def stat_on_local(self):
     wd_ = tempfile.mkdtemp(prefix='pydoop_', suffix=UNI_CHR)
     p_ = os.path.join(wd_, make_random_str())
     if hdfs.default_is_local():
         wd, p = wd_, p_
         host = "default"
     else:
         wd, p = ('file:%s' % _ for _ in (wd_, p_))
         host = ""
     fs = hdfs.hdfs(host, 0)
     with fs.open_file(p_, 'w') as fo:
         fo.write(make_random_str())
     info = fs.get_path_info(p_)
     fs.close()
     s = hdfs.path.stat(p)
     os_s = os.stat(p_)
     for n in dir(s):
         if n.startswith('st_'):
             try:
                 exp_v = getattr(os_s, n)
             except AttributeError:
                 try:
                     exp_v = info[self.NMAP[n]]
                 except KeyError:
                     continue
                 self.assertEqual(getattr(s, n), exp_v)
     self.__check_extra_args(s, info)
     self.__check_wrapper_funcs(p)
     hdfs.rmr(wd)

 def samefile_link(self):
     wd_ = tempfile.mkdtemp(prefix='pydoop_', suffix=UNI_CHR)
     wd = 'file:%s' % wd_
     link = os.path.join(wd_, make_random_str())
     os.symlink(wd_, link)
     self.assertTrue(hdfs.path.samefile('file:%s' % link, 'file:%s' % wd_))
     hdfs.rmr(wd)

 def good(self):
     base_path = make_random_str()
     for path in base_path, base_path + UNI_CHR:
         hdfs.dump("foo\n", path)
         self.assertTrue(hdfs.path.exists(path))
         hdfs.rmr(path)
         self.assertFalse(hdfs.path.exists(path))

 def _clean_up(*paths):
     for p in paths:
         try:
             log.debug("Removing path: %s", p)
             phdfs.rmr(p)
         except StandardError as e:
             log.warning("Error deleting path %s", p)
             log.exception(e)

 def realpath(self):
     wd_ = tempfile.mkdtemp(prefix='pydoop_', suffix=UNI_CHR)
     wd = 'file:%s' % wd_
     link = os.path.join(wd_, make_random_str())
     os.symlink(wd_, link)
     expected_path = 'file:%s' % os.path.realpath(wd_)
     self.assertEqual(hdfs.path.realpath('file:%s' % link), expected_path)
     hdfs.rmr(wd)

def cleanup(out_pathset):
  # clean-up job output
  for path in out_pathset:
    try:
      print >> sys.stderr, "Deleting output path", path
      phdfs.rmr(path)
    except StandardError as e:
      print >> sys.stderr, "Error!", str(e)

def _try_remove_hdfs_dir(path):
    try:
        phdfs.rmr(path)
        return True
    except StandardError as e:
        logger.error("Error while trying to remove directory %s", path)
        logger.exception(e)
    return False

 def __clean_wd(self):
   if self.remote_wd:
     try:
       self.logger.debug(
         "Removing temporary working directory %s", self.remote_wd
         )
       hdfs.rmr(self.remote_wd)
     except IOError:
       pass

 def realpath(self):
     wd_ = tempfile.mkdtemp(prefix='pydoop_', suffix=UNI_CHR)
     wd = 'file:%s' % wd_
     link = os.path.join(wd_, make_random_str())
     os.symlink(wd_, link)
     expected_path = ('file:%s%s' % ("/private", wd_)
                      if sys.platform == "darwin"
                      else 'file:%s' % wd_)
     self.assertEqual(hdfs.path.realpath('file:%s' % link), expected_path)
     hdfs.rmr(wd)

def delete_files(remote_basedir, retention):
    inodes = walk_remotely(remote_basedir)
    now = time.time()
    deleted_files = []
    for inode in inodes:
        if now - inode['last_mod'] > retention and inode['kind'] == 'file':
            LOGGER.debug("Deleting file {0}".format(inode['path']))
            hdfs.rmr(inode['path'])
            deleted_files.append(inode['path'])
    return deleted_files

 def runTest(self):
     path = make_random_str() + UNI_CHR
     hdfs.dump("foo\n", path)
     st = hdfs.path.stat(path)
     atime, mtime = [getattr(st, 'st_%stime' % _) for _ in 'am']
     new_atime, new_mtime = atime + 100, mtime + 200
     hdfs.path.utime(path, (new_atime, new_mtime))
     st = hdfs.path.stat(path)
     self.assertEqual(st.st_atime, new_atime)
     self.assertEqual(st.st_mtime, new_mtime)
     hdfs.rmr(path)

def _tear_down_flink_session(app_id):
    if not app_id:
        raise ValueError("_tear_down_flink_session: empty app id!")

    cmd = [ 'yarn', 'application', '-kill', app_id ]
    logger.info("Killing flink session with app id '%s'", app_id)
    logger.debug("Command: %s", cmd)
    subprocess.check_call(cmd)
    # clean up temporary yarn session files, if any
    path = ".flink/" + app_id
    if phdfs.path.exists(path):
        logger.info("Also removing the session's temporary files in %s", path)
        phdfs.rmr(path)

 def test_isdir(self):
     for path in self.path, self.u_path:
         self.assertFalse(hdfs.path.isdir(path))
         try:
             hdfs.dump("foo\n", path)
             self.assertFalse(hdfs.path.isdir(path))
             hdfs.rmr(path)
             hdfs.mkdir(path)
             self.assertTrue(hdfs.path.isdir(path))
         finally:
             try:
                 hdfs.rmr(path)
             except IOError:
                 pass

 def test_kind(self):
     for path in self.path, self.u_path:
         self.assertTrue(hdfs.path.kind(path) is None)
         try:
             hdfs.dump("foo\n", path)
             self.assertEqual('file', hdfs.path.kind(path))
             hdfs.rmr(path)
             hdfs.mkdir(path)
             self.assertEqual('directory', hdfs.path.kind(path))
         finally:
             try:
                 hdfs.rmr(path)
             except IOError:
                 pass

 def test_isdir(self):
   path = utils.make_random_str()
   self.assertFalse(hdfs.path.isdir(path))
   try:
     hdfs.dump("foo\n", path)
     self.assertFalse(hdfs.path.isdir(path))
     hdfs.rmr(path)
     hdfs.mkdir(path)
     self.assertTrue(hdfs.path.isdir(path))
   finally:
     try:
       hdfs.rmr(path)
     except IOError:
       pass

 def test_kind(self):
   path = utils.make_random_str()
   self.assertTrue(hdfs.path.kind(path) is None)
   try:
     hdfs.dump("foo\n", path)
     self.assertEqual('file', hdfs.path.kind(path))
     hdfs.rmr(path)
     hdfs.mkdir(path)
     self.assertEqual('directory', hdfs.path.kind(path))
   finally:
     try:
       hdfs.rmr(path)
     except IOError:
       pass

def main(argv):
  logger = logging.getLogger("main")
  logger.setLevel(logging.INFO)
  local_input = argv[1]
  with open(MR_SCRIPT) as f:
    pipes_code = pts.add_sys_path(f.read())
  runner = hadut.PipesRunner(prefix=PREFIX, logger=logger)
  runner.set_input(local_input, put=True)
  runner.set_exe(pipes_code)
  runner.run()
  res = runner.collect_output()
  runner.clean()
  hdfs.rmr(HDFS_WD)
  logger.info("checking results")
  expected_res = local_vc(local_input)
  logger.info(check(res, expected_res))

    def execute(self):
        """
        Execute workflow in dedicated directory
        """
        hdfs_output_dir = "workflow_output_{}".format(time.time())
        logger.debug("Setting up workflow")
        logger.debug("CWD: %s", os.getcwd())
        logger.debug("workflow output directory: %s", hdfs_output_dir)
        cmd = [ self._program ] + [ str(arg) for arg in  self._args ]
        cmd.append(self._input_dir)
        cmd.append(hdfs_output_dir)
        logger.debug("workflow command: %s", cmd)
        wf_logfile = os.path.abspath(GlobalConf['workflow_logfile'])
        logger.info("Executing worflow")
        logger.info("Writing workflow log to %s", wf_logfile)

        self._clear_caches()

        try:
            with open(wf_logfile, 'a') as f:
                logger.info("Starting workflow")
                start_time = time.time()
                retcode = subprocess.call(cmd, stdout=f, stderr=subprocess.STDOUT)
            end_time = time.time()
            run_time = end_time - start_time

            attempt_info = AttemptInfo(cmd, retcode, wf_logfile, run_time)

            if retcode == 0:
                logger.info("Workflow finished")
                logger.info("Attempt took %0.2f seconds", run_time)
                bcl, align = self._get_part_times_from_log(wf_logfile)
                attempt_info.bcl_secs = bcl
                attempt_info.align_secs = align
            else:
                logger.info("Workflow FAILED with exit code %s", retcode)
            return attempt_info
        finally:
            try:
                if phdfs.path.exists(hdfs_output_dir):
                    logger.debug("Removing workflow's temporary output directory %s", hdfs_output_dir)
                    phdfs.rmr(hdfs_output_dir)
            except StandardError as e:
                logger.error("Failed to clean up workflow's output directory  %s", hdfs_output_dir)
                logger.exception(e)

 def stat(self):
     if hdfs.default_is_local():
         return
     bn = '%s%s' % (make_random_str(), UNI_CHR)
     fn = '/user/%s/%s' % (DEFAULT_USER, bn)
     fs = hdfs.hdfs("default", 0)
     p = "hdfs://%s:%s%s" % (fs.host, fs.port, fn)
     with fs.open_file(fn, 'w') as fo:
         fo.write(make_random_str())
     info = fs.get_path_info(fn)
     fs.close()
     s = hdfs.path.stat(p)
     for n1, n2 in self.NMAP.iteritems():
         attr = getattr(s, n1, None)
         self.assertFalse(attr is None)
         self.assertEqual(attr, info[n2])
     self.__check_extra_args(s, info)
     self.__check_wrapper_funcs(p)
     hdfs.rmr(p)

 def run(self):
     exit_code = 1
     with tempfile.NamedTemporaryFile() as f:
         self.log.debug("opened scratch MR job input file %s", f.name)
         # We write the files to be compressed to a temporary file.  Later we'll re-read
         # this temporary file to rename the files as well.  I've opted not to keep the
         # table in memory in the hope of scaling better to jobs with a large number of
         # files (we reduce memory requirements).
         num_files = self.__write_mr_input(f)
         f.flush()
         self.log.debug("Finished writing temp input file")
         input_filename = tempfile.mktemp(dir=os.path.dirname(self.output_path), prefix="dist_txt_zipper_input")
         tmpfile_uri = "file://%s" % f.name
         try:
             self.log.debug("copying input from %s to %s", tmpfile_uri, input_filename)
             hdfs.cp(tmpfile_uri, input_filename)
             self.log.info("Run analyzed.  Launching distributed job")
             # launch mr task
             pydoop_args = \
                 [ 'script', '--num-reducers', '0','--kv-separator', '',
                   '-Dmapred.map.tasks=%d' % num_files,
                   '-Dmapred.input.format.class=org.apache.hadoop.mapred.lib.NLineInputFormat',
                   '-Dmapred.line.input.format.linespermap=1',
                   '-Dmapred.output.compress=true',
                   '-Dmapred.output.compression.codec=%s' % 'org.apache.hadoop.io.compress.GzipCodec',
                   text_zipper_mr.__file__,
                   input_filename,
                   self.output_path]
             self.log.debug("pydoop_args: %s", pydoop_args)
             self.log.info("Compressing %s files", num_files)
             pydoop_app.main(pydoop_args)
             self.log.info("Distributed job complete")
             self.rename_compressed_files(f)
             self.log.info("finished")
             exit_code = 0
         finally:
             try:
                 self.log.debug("Removing temporary input file %s", input_filename)
                 hdfs.rmr(input_filename)
             except IOError as e:
                 self.log.warning("Problem cleaning up.  Error deleting temporary input file %s", input_filename)
                 self.log.exception(str(e))
         return exit_code