def __init__(self, env):
   self.env = env
   self.id = str(uuid.uuid4())
   self.curl = utils.curl(self.env.url('/debug/querylog'), background=True, stdout=open(self.path, 'w'))
   self.curl_full = utils.curl(self.env.url('/debug/querylog?full=true'), background=True, stdout=open(self.path_full, 'w'))
   self.tailer = framework.Tailer(self.path, sleep=0.02)
   self.tailer_full = framework.Tailer(self.path_full, sleep=0.02)

  def wipe(self):
    import utils  # pylint: disable=g-import-not-at-top

    utils.curl('http://' + self.server_addr + '/v1/kv/global/keyspaces?recurse',
               request='DELETE')
    for cell in ['test_nj', 'test_ny', 'test_ca']:
      utils.curl('http://' + self.server_addr + '/v1/kv/' + cell + '?recurse',
                 request='DELETE')

  def wait_until_up(self):
    import utils  # pylint: disable=g-import-not-at-top

    # Wait for global cluster to come up.
    # We create a dummy directory using v2 API, won't be visible to v3.
    utils.curl(
        self.api_url + '/keys/test', request='PUT',
        data='dir=true', retry_timeout=10)

 def __init__(self, env):
     self.env = env
     self.id = str(uuid.uuid4())
     self.curl = utils.curl(self.env.url("/debug/querylog"), background=True, stdout=open(self.path, "w"))
     self.curl_full = utils.curl(
         self.env.url("/debug/querylog?full=true"), background=True, stdout=open(self.path_full, "w")
     )
     time.sleep(0.3)
     self.tailer = framework.Tailer(open(self.path), sleep=0.02)
     self.tailer_full = framework.Tailer(open(self.path_full), sleep=0.02)

    def sync_incremental_res(self):
        mode = 'increment' if self._is_art else 'full'
        can_sync_inc_res = False
        for module in self._all_modules:
            finder = GradleDirectoryFinder(module, self._project_info[module]['path'], self._cache_dir)
            fpath = finder.get_dst_res_pack_path(module)
            sync_status = finder.get_sync_file_path()

            if not os.path.exists(sync_status):
                self.debug('{} has no need to sync inc res pack.'.format(module))
                continue

            if not can_sync_inc_res and self._is_art:
                self.check_base_res_exist()
                self.push_full_res_pack()
                can_sync_inc_res = True

            self.debug('start to sync {} incremental res pack...'.format(module))
            self.debug('{} pack size: {}kb'.format(module, os.path.getsize(fpath) / 1000))
            with open(fpath, 'rb') as fp:
                url = 'http://127.0.0.1:{}/pushResource?mode={}&bundleId={}'.format(self._port, mode, 'base-res')
                self.debug('pushres: ' + url)
                result, err, code = curl(url, body=fp.read())
                if code != 0:
                    raise FreelineException('sync incremental respack failed', err.message)

                android_tools.clean_res_build_job_flag(finder.get_res_build_job_path())
                self.debug('sync {} incremental res pack finished'.format(module))

  def setUp(self):
    utils.wait_procs([self.tablet.init_mysql()])
    self.tablet.mquery(
        '', ['create database vt_test_keyspace', 'set global read_only = off'])

    self.mysql_conn, mcu = self.tablet.connect('vt_test_keyspace')
    with open(
        os.path.join(self.vttop, 'test', 'test_data', 'test_schema.sql')) as f:
      self.clean_sqls = []
      self.init_sqls = []
      clean_mode = False
      for line in f:
        line = line.rstrip()
        if line == '# clean':
          clean_mode = True
        if line=='' or line.startswith('#'):
          continue
        if clean_mode:
          self.clean_sqls.append(line)
        else:
          self.init_sqls.append(line)
      try:
        for line in self.init_sqls:
          mcu.execute(line, {})
      finally:
        mcu.close()

    customrules = os.path.join(environment.tmproot, 'customrules.json')
    schema_override = os.path.join(environment.tmproot, 'schema_override.json')
    self.create_schema_override(schema_override)
    table_acl_config = os.path.join(
        environment.vttop, 'test', 'test_data', 'table_acl_config.json')

    environment.topo_server().setup()
    self.create_customrules(customrules);
    utils.run_vtctl('CreateKeyspace -force test_keyspace')
    self.tablet.init_tablet('master', 'test_keyspace', '0')
    if environment.topo_server().flavor() == 'zookeeper':
      self.tablet.start_vttablet(
          memcache=self.memcache,
          zkcustomrules='/zk/test_ca/config/customrules/testrules',
          schema_override=schema_override,
          table_acl_config=table_acl_config,
      )
    else:
      self.tablet.start_vttablet(
          memcache=self.memcache,
          filecustomrules=customrules,
          schema_override=schema_override,
          table_acl_config=table_acl_config,
      )
    self.conn = self.connect()
    self.txlogger = utils.curl(
        self.url('/debug/txlog'), background=True,
        stdout=open(self.txlog_file, 'w'))
    self.txlog = framework.Tailer(self.txlog_file, flush=self.tablet.flush)
    self.log = framework.Tailer(
        os.path.join(environment.vtlogroot, 'vttablet.INFO'),
        flush=self.tablet.flush)
    self.querylog = Querylog(self)

    def scan_device_port(self):
        port = 0
        apktime_path = self._get_apktime_path()
        self.debug("apktime path: " + apktime_path)
        sync_value = get_sync_value(apktime_path, self._cache_dir)
        self.debug('your local sync value is: {}'.format(sync_value))
        uuid = get_apk_created_ticket(apktime_path)
        self.debug('your local uuid value is: {}'.format(uuid))

        for i in range(0, 10):
            cexec([self._adb, 'forward', 'tcp:{}'.format(41128 + i), 'tcp:{}'.format(41128 + i)], callback=None)
            url = 'http://127.0.0.1:{}/checkSync?sync={}&uuid={}'.format(41128 + i, sync_value, uuid)
            result, err, code = curl(url)
            if code == 0 and result is not None:
                result = int(result)
                self.debug('server result is {}'.format(result))
                if result == 0:
                    self.debug('check sync value failed, maybe you need a clean build.')
                    from exceptions import CheckSyncStateException
                    raise CheckSyncStateException('check sync value failed, maybe you need a clean build.',
                                                  'NO CAUSE')
                elif result == -1:
                    continue
                else:
                    port = 41128 + i
                    break

        for i in range(0, 10):
            if (41128 + i) != port:
                cexec([self._adb, 'forward', '--remove', 'tcp:{}'.format(41128 + i)], callback=None)

        return port

    def scan_device_port(self, sync_value, uuid):
        port = 0

        for i in range(0, 10):
            cexec([self._adb, 'forward', 'tcp:{}'.format(PORT_START + i), 'tcp:{}'.format(PORT_START + i)],
                  callback=None)
            url = 'http://127.0.0.1:{}/checkSync?sync={}&uuid={}'.format(PORT_START + i, sync_value, uuid)
            result, err, code = curl(url)
            if code == 0 and result is not None:
                result = int(result)
                self.debug('server result is {}'.format(result))
                if result == 0:
                    self.debug('check sync value failed, maybe you need a clean build.')
                    from exceptions import CheckSyncStateException
                    raise CheckSyncStateException('check sync value failed, maybe you need a clean build.',
                                                  'NO CAUSE')
                elif result == -1:
                    continue
                else:
                    port = PORT_START + i
                    break

        for i in range(0, 10):
            if (PORT_START + i) != port:
                cexec([self._adb, 'forward', '--remove', 'tcp:{}'.format(PORT_START + i)], callback=None)

        return port

  def setUp(self):
    environment.topo_server_setup()

    utils.wait_procs([self.tablet.init_mysql()])
    self.tablet.mquery("", ["create database vt_test_keyspace", "set global read_only = off"])

    self.mysql_conn, mcu = self.tablet.connect('vt_test_keyspace')
    self.clean_sqls = []
    self.init_sqls = []
    clean_mode = False
    with open(os.path.join(self.vttop, "test", "test_data", "test_schema.sql")) as f:
      for line in f:
        line = line.rstrip()
        if line == "# clean":
          clean_mode = True
        if line=='' or line.startswith("#"):
          continue
        if clean_mode:
          self.clean_sqls.append(line)
        else:
          self.init_sqls.append(line)
    try:
      for line in self.init_sqls:
        mcu.execute(line, {})
    finally:
      mcu.close()

    utils.run_vtctl('CreateKeyspace -force test_keyspace')
    self.tablet.init_tablet('master', 'test_keyspace', '0')

    customrules = os.path.join(environment.tmproot, 'customrules.json')
    self.create_customrules(customrules)
    schema_override = os.path.join(environment.tmproot, 'schema_override.json')
    self.create_schema_override(schema_override)
    table_acl_config = os.path.join(environment.vttop, 'test', 'test_data', 'table_acl_config.json')
    self.tablet.start_vttablet(
            memcache=self.memcache,
            customrules=customrules,
            schema_override=schema_override,
            table_acl_config=table_acl_config,
            auth=True,
    )

    # FIXME(szopa): This is necessary here only because of a bug that
    # makes the qs reload its config only after an action.
    utils.run_vtctl('Ping ' + self.tablet.tablet_alias)

    for i in range(30):
      try:
        self.conn = self.connect()
        self.txlogger = utils.curl(self.url('/debug/txlog'), background=True, stdout=open(self.txlog_file, 'w'))
        self.txlog = framework.Tailer(open(self.txlog_file), flush=self.tablet.flush)
        self.log = framework.Tailer(open(os.path.join(environment.vtlogroot, 'vttablet.INFO')), flush=self.tablet.flush)
        break
      except dbexceptions.OperationalError:
        if i == 29:
          raise
        time.sleep(1)
    self.postSetup()

  def setup(self, add_bad_host=False):
    import utils

    for cell in ['global', 'test_ca', 'test_nj', 'test_ny']:
      self.clusters[cell] = EtcdCluster(cell)

    # Wait for global cluster to come up.
    utils.curl(
        self.clusters['global'].api_url + '/keys/vt', request='PUT',
        data='dir=true', retry_timeout=10)

    # Add entries in global cell list.
    for cell, cluster in self.clusters.iteritems():
      if cell != 'global':
        utils.curl(
            '%s/keys/vt/cells/%s' %
            (self.clusters['global'].api_url, cell), request='PUT',
            data='value=http://' + cluster.client_addr)

 def sync_incremental_native(self):
     if self._is_need_sync_native():
         self.debug('start to sync native file...')
         native_zip_path = get_sync_native_file_path(self._config['build_cache_dir'])
         with open(native_zip_path, "rb") as fp:
             url = "http://127.0.0.1:{}/pushNative?restart".format(self._port)
             self.debug("pushNative: "+url)
             result, err, code = curl(url, body=fp.read())
             self.debug("code: {}".format(code))

  def setUp(self):
    utils.wait_procs([self.tablet.init_mysql()])
    self.tablet.mquery("", ["create database vt_test_keyspace", "set global read_only = off"])

    self.mysql_conn, mcu = self.tablet.connect('vt_test_keyspace')
    with open(os.path.join(self.vttop, "test", "test_data", "test_schema.sql")) as f:
      self.clean_sqls = []
      self.init_sqls = []
      clean_mode = False
      for line in f:
        line = line.rstrip()
        if line == "# clean":
          clean_mode = True
        if line=='' or line.startswith("#"):
          continue
        if clean_mode:
          self.clean_sqls.append(line)
        else:
          self.init_sqls.append(line)
      try:
        for line in self.init_sqls:
          mcu.execute(line, {})
      finally:
        mcu.close()

    customrules = os.path.join(environment.tmproot, 'customrules.json')
    self.create_customrules(customrules)
    schema_override = os.path.join(environment.tmproot, 'schema_override.json')
    self.create_schema_override(schema_override)
    table_acl_config = os.path.join(environment.vttop, 'test', 'test_data', 'table_acl_config.json')

    if self.env == 'vttablet':
      environment.topo_server().setup()
      utils.run_vtctl('CreateKeyspace -force test_keyspace')
      self.tablet.init_tablet('master', 'test_keyspace', '0')
      self.tablet.start_vttablet(
              memcache=self.memcache,
              customrules=customrules,
              schema_override=schema_override,
              table_acl_config=table_acl_config,
              auth=True,
      )
    else:
      self.tablet.start_vtocc(
              memcache=self.memcache,
              customrules=customrules,
              schema_override=schema_override,
              table_acl_config=table_acl_config,
              auth=True,
              keyspace="test_keyspace", shard="0",
      )
    self.conn = self.connect()
    self.txlogger = utils.curl(self.url('/debug/txlog'), background=True, stdout=open(self.txlog_file, 'w'))
    self.txlog = framework.Tailer(self.txlog_file, flush=self.tablet.flush)
    self.log = framework.Tailer(os.path.join(environment.vtlogroot, '%s.INFO' % self.env), flush=self.tablet.flush)
    self.querylog = Querylog(self)

 def check_base_res_exist(self):
     url = 'http://127.0.0.1:{}/checkResource'.format(self._port)
     self.debug('checkresource: ' + url)
     result, err, code = curl(url)
     if code != 0:
         raise FreelineException('check base res failed', err.message)
     if int(result) == 0:
         self.debug('base resource not exists, need to sync full resource pack first')
         self._is_need_sync_base_res = True
     else:
         self.debug('base resource exists, there is no need to sync full resource pack')

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

    self.port_base = environment.reserve_ports(5)
    self.server_addr = 'localhost:%d' % (self.port_base + 1)

    # Write our config file.
    self.config_file = os.path.join(environment.vtdataroot, 'consul.json')
    config = {
        'ports': {
            'dns': self.port_base,
            'http': self.port_base + 1,
            'rpc': self.port_base + 2,
            'serf_lan': self.port_base + 3,
            'serf_wan': self.port_base + 4,
        },
    }
    with open(self.config_file, 'w') as fd:
      fd.write(json.dumps(config))

    log_base = os.path.join(environment.vtlogroot, 'consul')
    self.proc = utils.run_bg([
        'consul', 'agent',
        '-dev',
        '-config-file', self.config_file],
                             stdout=open(log_base + '.stdout', 'a'),
                             stderr=open(log_base + '.stderr', 'a'))

    # Wait until the daemon is ready.
    utils.curl(
        'http://' + self.server_addr + '/v1/kv/?keys', retry_timeout=10)

    # Create the cell configurations using 'vtctl AddCellInfo'
    for cell in ['test_nj', 'test_ny', 'test_ca']:
      utils.run_vtctl_vtctl(['AddCellInfo',
                             '-root', cell,
                             '-server_address', self.server_addr,
                             cell])

 def sync_incremental_dex(self):
     dex_path = android_tools.get_incremental_dex_path(self._cache_dir)
     if os.path.exists(dex_path):
         self.debug('start to sync incremental dex...')
         with open(dex_path, 'rb') as fp:
             url = 'http://127.0.0.1:{}/pushDex'.format(self._port)
             self.debug('pushdex: ' + url)
             result, err, code = curl(url, body=fp.read())
             if code != 0:
                 from exceptions import FreelineException
                 raise FreelineException('sync incremental dex failed.', err.message)
     else:
         self.debug('no {} exists.'.format(dex_path))

 def sync_state(self, is_need_restart):
     if self._is_need_sync_dex() or self._is_need_sync_res():
         self.debug('start to sync close longlink...')
         restart_char = 'restart' if is_need_restart else 'no'
         update_last_sync_ticket(self._cache_dir)
         url = 'http://127.0.0.1:{}/closeLongLink?{}&lastSync={}'.format(self._port, restart_char,
                                                                         get_last_sync_ticket(self._cache_dir))
         self.debug('closeLongLink: ' + url)
         result, err, code = curl(url)
         # self.wake_up()
         if code != 0:
             rollback_last_sync_ticket(self._cache_dir)
             from exceptions import FreelineException
             raise FreelineException('sync state failed.', err.message)

  def wipe(self):
    import utils

    for cell, cluster in self.clusters.iteritems():
      if cell == 'global':
        utils.curl(
            cluster.api_url + '/keys/vt/keyspaces?recursive=true',
            request='DELETE')
      else:
        utils.curl(
            cluster.api_url + '/keys/vt/ns?recursive=true', request='DELETE')
        utils.curl(
            cluster.api_url + '/keys/vt/tablets?recursive=true',
            request='DELETE')
        utils.curl(
            cluster.api_url + '/keys/vt/replication?recursive=true',
            request='DELETE')

 def push_full_res_pack(self):
     if self._is_need_sync_base_res:
         full_pack_path = get_base_resource_path(self._cache_dir)
         if os.path.exists(full_pack_path):
             self.debug('start to sync full resource pack...')
             self.debug('full pack size: {}kb'.format(os.path.getsize(full_pack_path) / 1000))
             with open(full_pack_path, 'rb') as fp:
                 url = 'http://127.0.0.1:{}/pushFullResourcePack'.format(self._port)
                 self.debug('pushfullpack: ' + url)
                 result, err, code = curl(url, body=fp.read())
                 if code != 0:
                     raise FreelineException('push full res pack failed', err.message)
         else:
             raise FreelineException('You may need a clean build.',
                                     'full resource pack not found: {}'.format(full_pack_path))

 def sync_incremental_dex(self):
     # dex_path = android_tools.get_incremental_dex_path(self._cache_dir)
     dex_dir = android_tools.get_incremental_dex_dir(self._cache_dir)
     if os.path.isdir(dex_dir):
         dexes = [fn for fn in os.listdir(dex_dir) if fn.endswith('.dex')]
         if len(dexes) > 0:
             self.debug('start to sync incremental dex...')
             for dex_name in dexes:
                 dex_path = os.path.join(dex_dir, dex_name)
                 with open(dex_path, 'rb') as fp:
                     url = 'http://127.0.0.1:{}/pushDex?dexName={}'.format(self._port, dex_name.replace('.dex', ''))
                     self.debug('pushdex: ' + url)
                     self.debug('dex path: {}'.format(dex_path))
                     result, err, code = curl(url, body=fp.read())
                     if code != 0:
                         from exceptions import FreelineException
                         raise FreelineException('sync incremental dex failed.', err.message)
         else:
             self.debug('no incremental dexes in .'.format(dex_dir))

    def scan_device_port(self, sync_value, uuid):
        port = 0

        for i in range(0, 10):
            cexec([self._adb, 'forward', 'tcp:{}'.format(PORT_START + i), 'tcp:{}'.format(PORT_START + i)],
                  callback=None)
            url = 'http://127.0.0.1:{}/getSyncTicket'.format(PORT_START + i)
            self.debug('url===='+url)
            result, err, code = curl(url)
            if code == 0 and result is not None:
                try:
                    result = json.loads(result.replace("'", '"'))
                    self.debug(result)
                    if result["apkBuildFlag"] == uuid:
                        port = PORT_START + i
                        break
                    elif result["apkBuildFlag"] is not None:
                        self.debug('apkBuildFlag: {} does not match uuid: {}'.format(result["apkBuildFlag"], uuid))
                        port = -1
                        break
                        
                except Exception, e:
                    pass