def tearDown(self):
   try:
     self.server.close_start()
     file_path.rmtree(self.tempdir)
     self.server.close_end()
   finally:
     super(IsolateServerStorageSmokeTest, self).tearDown()

def archive(isolate_server, script):
  """Archives the tool and return the sha-1."""
  base_script = os.path.basename(script)
  isolate = {
    'variables': {
      'command': ['python', base_script],
      'files': [base_script],
    },
  }
  tempdir = tempfile.mkdtemp(prefix=u'run_on_bots')
  try:
    isolate_file = os.path.join(tempdir, 'tool.isolate')
    isolated_file = os.path.join(tempdir, 'tool.isolated')
    with open(isolate_file, 'wb') as f:
      f.write(str(isolate))
    shutil.copyfile(script, os.path.join(tempdir, base_script))
    cmd = [
      sys.executable, 'isolate.py', 'archive',
      '--isolate-server', isolate_server,
      '-i', isolate_file,
      '-s', isolated_file,
    ]
    return subprocess.check_output(cmd, cwd=ROOT_DIR).split()[0]
  finally:
    file_path.rmtree(tempdir)

 def tearDown(self):
   try:
     self.server.close_start()
     file_path.rmtree(self.tempdir)
     self.server.close_end()
   finally:
     super(RunIsolatedTest, self).tearDown()

 def task_collect(self, task_id):
   """Collects the results for a task."""
   h, tmp = tempfile.mkstemp(prefix='swarming_smoke_test', suffix='.json')
   os.close(h)
   try:
     tmpdir = tempfile.mkdtemp(prefix='swarming_smoke_test')
     try:
       # swarming.py collect will return the exit code of the task.
       args = [
         '--task-summary-json', tmp, task_id, '--task-output-dir', tmpdir,
         '--timeout', '20', '--perf',
       ]
       self._run('collect', args)
       with open(tmp, 'rb') as f:
         content = f.read()
       try:
         summary = json.loads(content)
       except ValueError:
         print >> sys.stderr, 'Bad json:\n%s' % content
         raise
       outputs = {}
       for root, _, files in os.walk(tmpdir):
         for i in files:
           p = os.path.join(root, i)
           name = p[len(tmpdir)+1:]
           with open(p, 'rb') as f:
             outputs[name] = f.read()
       return summary, outputs
     finally:
       file_path.rmtree(tmpdir)
   finally:
     os.remove(tmp)

  def test_get_swarming_bot_zip(self):
    zipped_code = bot_code.get_swarming_bot_zip('http://localhost')
    # Ensure the zip is valid and all the expected files are present.
    with zipfile.ZipFile(StringIO.StringIO(zipped_code), 'r') as zip_file:
      for i in bot_archive.FILES:
        with zip_file.open(i) as f:
          content = f.read()
          if os.path.basename(i) != '__init__.py':
            self.assertTrue(content, i)

    temp_dir = tempfile.mkdtemp(prefix='swarming')
    try:
      # Try running the bot and ensure it can import the required files. (It
      # would crash if it failed to import them).
      bot_path = os.path.join(temp_dir, 'swarming_bot.zip')
      with open(bot_path, 'wb') as f:
        f.write(zipped_code)
      proc = subprocess.Popen(
          [sys.executable, bot_path, 'start_bot', '-h'],
          cwd=temp_dir,
          stdout=subprocess.PIPE,
          stderr=subprocess.STDOUT)
      out = proc.communicate()[0]
      self.assertEqual(0, proc.returncode, out)
    finally:
      file_path.rmtree(temp_dir)

 def test_rmtree_unicode(self):
     subdir = os.path.join(self.tempdir, "hi")
     fs.mkdir(subdir)
     filepath = os.path.join(subdir, u"\u0627\u0644\u0635\u064A\u0646\u064A\u0629")
     with fs.open(filepath, "wb") as f:
         f.write("hi")
     # In particular, it fails when the input argument is a str.
     file_path.rmtree(str(subdir))

 def tearDown(self):
   for dirpath, dirnames, filenames in os.walk(self.tempdir, topdown=True):
     for filename in filenames:
       file_path.set_read_only(os.path.join(dirpath, filename), False)
     for dirname in dirnames:
       file_path.set_read_only(os.path.join(dirpath, dirname), False)
   file_path.rmtree(self.tempdir)
   super(RunIsolatedTestBase, self).tearDown()

 def tearDown(self):
     try:
         if self._tempdir:
             file_path.rmtree(self._tempdir)
         if not self.has_failed():
             self._check_output("", "")
     finally:
         super(NetTestCase, self).tearDown()

 def tearDown(self):
   os.chdir(test_env_bot_code.BOT_DIR)
   try:
     file_path.rmtree(self.root_dir)
   except OSError:
     print >> sys.stderr, 'Failed to delete %s' % self.root_dir
   finally:
     super(TestTaskRunnerBase, self).tearDown()

 def test_rmtree_unicode(self):
   subdir = os.path.join(self.tempdir, 'hi')
   os.mkdir(subdir)
   filepath = os.path.join(
       subdir, u'\u0627\u0644\u0635\u064A\u0646\u064A\u0629')
   with open(filepath, 'wb') as f:
     f.write('hi')
   # In particular, it fails when the input argument is a str.
   file_path.rmtree(str(subdir))

 def test_undeleteable_chmod(self):
   # Create a file and a directory with an empty ACL. Then try to delete it.
   dirpath = os.path.join(self.tempdir, 'd')
   filepath = os.path.join(dirpath, 'f')
   os.mkdir(dirpath)
   with open(filepath, 'w') as f:
     f.write('hi')
   os.chmod(filepath, 0)
   os.chmod(dirpath, 0)
   file_path.rmtree(dirpath)

 def tearDown(self):
     try:
         if self._tempdir:
             for dirpath, dirnames, filenames in fs.walk(self._tempdir, topdown=True):
                 for filename in filenames:
                     file_path.set_read_only(os.path.join(dirpath, filename), False)
                 for dirname in dirnames:
                     file_path.set_read_only(os.path.join(dirpath, dirname), False)
             file_path.rmtree(self._tempdir)
     finally:
         super(FilePathTest, self).tearDown()

def main():
  tools.disable_buffering()
  parser = optparse.OptionParser()
  parser.add_option('-s', '--isolated', help='.isolated file to profile with.')
  parser.add_option('--largest_files', type='int',
                    help='If this is set, instead of compressing all the '
                    'files, only the large n files will be compressed')
  options, args = parser.parse_args()

  if args:
    parser.error('Unknown args passed in; %s' % args)
  if not options.isolated:
    parser.error('The .isolated file must be given.')

  temp_dir = None
  try:
    temp_dir = tempfile.mkdtemp(prefix=u'zip_profiler')

    # Create a directory of the required files
    subprocess.check_call([os.path.join(ROOT_DIR, 'isolate.py'),
                           'remap',
                           '-s', options.isolated,
                           '--outdir', temp_dir])

    file_set = tree_files(temp_dir)

    if options.largest_files:
      sorted_by_size = sorted(file_set.iteritems(),  key=lambda x: x[1],
                              reverse=True)
      files_to_compress = sorted_by_size[:options.largest_files]

      for filename, size in files_to_compress:
        print('Compressing %s, uncompressed size %d' % (filename, size))

        profile_compress('zlib', zlib.compressobj, range(10), zip_file,
                         filename)
        profile_compress('bz2', bz2.BZ2Compressor, range(1, 10), zip_file,
                         filename)
    else:
      print('Number of files: %s' % len(file_set))
      print('Total size: %s' % sum(file_set.itervalues()))

      # Profile!
      profile_compress('zlib', zlib.compressobj, range(10), zip_directory,
                       temp_dir)
      profile_compress('bz2', bz2.BZ2Compressor, range(1, 10), zip_directory,
                       temp_dir)
  finally:
    file_path.rmtree(temp_dir)

def CMDrun(parser, args):
    """Runs the test executable in an isolated (temporary) directory.

  All the dependencies are mapped into the temporary directory and the
  directory is cleaned up after the target exits.

  Argument processing stops at -- and these arguments are appended to the
  command line of the target to run. For example, use:
    isolate.py run --isolated foo.isolated -- --gtest_filter=Foo.Bar
  """
    add_isolate_options(parser)
    add_skip_refresh_option(parser)
    options, args = parser.parse_args(args)
    process_isolate_options(parser, options, require_isolated=False)
    complete_state = load_complete_state(options, os.getcwd(), None, options.skip_refresh)
    cmd = complete_state.saved_state.command + args
    if not cmd:
        raise ExecutionError("No command to run.")
    cmd = tools.fix_python_path(cmd)

    outdir = run_isolated.make_temp_dir(u"isolate-%s" % datetime.date.today(), os.path.dirname(complete_state.root_dir))
    try:
        # TODO(maruel): Use run_isolated.run_tha_test().
        cwd = create_isolate_tree(
            outdir,
            complete_state.root_dir,
            complete_state.saved_state.files,
            complete_state.saved_state.relative_cwd,
            complete_state.saved_state.read_only,
        )
        file_path.ensure_command_has_abs_path(cmd, cwd)
        logging.info("Running %s, cwd=%s" % (cmd, cwd))
        try:
            result = subprocess.call(cmd, cwd=cwd)
        except OSError:
            sys.stderr.write(
                "Failed to executed the command; executable is missing, maybe you\n"
                "forgot to map it in the .isolate file?\n  %s\n  in %s\n" % (" ".join(cmd), cwd)
            )
            result = 1
    finally:
        file_path.rmtree(outdir)

    if complete_state.isolated_filepath:
        complete_state.save_files()
    return result

 def _run_isolated(self, hello_world, name, args, expected_summary,
     expected_files):
   # Shared code for all test_isolated_* test cases.
   tmpdir = tempfile.mkdtemp(prefix='swarming_smoke')
   try:
     isolate_path = os.path.join(tmpdir, 'i.isolate')
     isolated_path = os.path.join(tmpdir, 'i.isolated')
     with open(isolate_path, 'wb') as f:
       json.dump(ISOLATE_HELLO_WORLD, f)
     with open(os.path.join(tmpdir, 'hello_world.py'), 'wb') as f:
       f.write(hello_world)
     isolated_hash = self.client.isolate(isolate_path, isolated_path)
     task_id = self.client.task_trigger_isolated(
         name, isolated_hash, extra=args)
     actual_summary, actual_files = self.client.task_collect(task_id)
     self.assertResults(expected_summary, actual_summary)
     actual_files.pop('summary.json')
     self.assertEqual(expected_files, actual_files)
   finally:
     file_path.rmtree(tmpdir)

        def test_native_case_alternate_datastream(self):
            # Create the file manually, since tempfile doesn't support ADS.
            tempdir = unicode(tempfile.mkdtemp(prefix=u"trace_inputs"))
            try:
                tempdir = file_path.get_native_path_case(tempdir)
                basename = "foo.txt"
                filename = basename + ":Zone.Identifier"
                filepath = os.path.join(tempdir, filename)
                open(filepath, "w").close()
                self.assertEqual(filepath, file_path.get_native_path_case(filepath))
                data_suffix = ":$DATA"
                self.assertEqual(filepath + data_suffix, file_path.get_native_path_case(filepath + data_suffix))

                open(filepath + "$DATA", "w").close()
                self.assertEqual(filepath + data_suffix, file_path.get_native_path_case(filepath + data_suffix))
                # Ensure the ADS weren't created as separate file. You love NTFS, don't
                # you?
                self.assertEqual([basename], fs.listdir(tempdir))
            finally:
                file_path.rmtree(tempdir)

def archive_isolated_triggers(isolate_server, tree_isolated, tests):
  """Creates and archives all the .isolated files for the tests at once.

  Archiving them in one batch is faster than archiving each file individually.
  Also the .isolated files can be reused across OSes, reducing the amount of
  I/O.

  Returns:
    list of (test, sha1) tuples.
  """
  logging.info('archive_isolated_triggers(%s, %s)', tree_isolated, tests)
  tempdir = tempfile.mkdtemp(prefix=u'run_swarming_tests_on_swarming_')
  try:
    isolateds = []
    for test in tests:
      test_name = os.path.basename(test)
      # Creates a manual .isolated file. See
      # https://code.google.com/p/swarming/wiki/IsolatedDesign for more details.
      isolated = {
        'algo': 'sha-1',
        'command': ['python', test],
        'includes': [tree_isolated],
        'read_only': 0,
        'version': '1.4',
      }
      v = os.path.join(tempdir, test_name + '.isolated')
      tools.write_json(v, isolated, True)
      isolateds.append(v)
    cmd = [
        'isolateserver.py', 'archive', '--isolate-server', isolate_server,
    ] + isolateds
    if logging.getLogger().isEnabledFor(logging.INFO):
      cmd.append('--verbose')
    items = [i.split() for i in check_output(cmd).splitlines()]
    assert len(items) == len(tests)
    assert all(
        items[i][1].endswith(os.path.basename(tests[i]) + '.isolated')
        for i in xrange(len(tests)))
    return zip(tests, [i[0] for i in items])
  finally:
    file_path.rmtree(tempdir)

    def test_rmtree_win(self):
      # Mock our sleep for faster test case execution.
      sleeps = []
      self.mock(time, 'sleep', sleeps.append)
      self.mock(sys, 'stderr', StringIO.StringIO())

      # Open a child process, so the file is locked.
      subdir = os.path.join(self.tempdir, 'to_be_deleted')
      fs.mkdir(subdir)
      script = 'import time; open(\'a\', \'w\'); time.sleep(60)'
      proc = subprocess.Popen([sys.executable, '-c', script], cwd=subdir)
      try:
        # Wait until the file exist.
        while not fs.isfile(os.path.join(subdir, 'a')):
          self.assertEqual(None, proc.poll())
        file_path.rmtree(subdir)
        self.assertEqual([2, 4, 2], sleeps)
        # sys.stderr.getvalue() would return a fair amount of output but it is
        # not completely deterministic so we're not testing it here.
      finally:
        proc.wait()

 def test_undeleteable_owner(self):
   # Create a file and a directory with an empty ACL. Then try to delete it.
   dirpath = os.path.join(self.tempdir, 'd')
   filepath = os.path.join(dirpath, 'f')
   os.mkdir(dirpath)
   with open(filepath, 'w') as f:
     f.write('hi')
   import win32security
   user, _domain, _type = win32security.LookupAccountName(
       '', getpass.getuser())
   sd = win32security.SECURITY_DESCRIPTOR()
   sd.Initialize()
   sd.SetSecurityDescriptorOwner(user, False)
   # Create an empty DACL, which removes all rights.
   dacl = win32security.ACL()
   dacl.Initialize()
   sd.SetSecurityDescriptorDacl(1, dacl, 0)
   win32security.SetFileSecurity(
       fs.extend(filepath), win32security.DACL_SECURITY_INFORMATION, sd)
   win32security.SetFileSecurity(
       fs.extend(dirpath), win32security.DACL_SECURITY_INFORMATION, sd)
   file_path.rmtree(dirpath)

def gen_isolated(isolate, script, includes=None):
  """Archives a script to `isolate` server."""
  tmp = tempfile.mkdtemp(prefix='swarming_smoke')
  data = {
    'variables': {
      'command': ['python', '-u', 'script.py'],
      'files': ['script.py'],
    },
  }
  try:
    with open(os.path.join(tmp, 'script.py'), 'wb') as f:
      f.write(script)
    path = os.path.join(tmp, 'script.isolate')
    with open(path, 'wb') as f:
      # This file is actually python but it's #closeenough.
      json.dump(data, f, sort_keys=True, separators=(',', ':'))
    isolated = os.path.join(tmp, 'script.isolated')
    cmd = [
      os.path.join(CLIENT_DIR, 'isolate.py'), 'archive',
      '-I', isolate, '-i', path, '-s', isolated,
    ]
    out = subprocess.check_output(cmd)
    if includes:
      # Mangle the .isolated to include another one. A bit hacky but works well.
      # In practice, we'd need to add a --include flag to isolate.py archive or
      # something.
      with open(isolated, 'rb') as f:
        data = json.load(f)
      data['includes'] = includes
      with open(isolated, 'wb') as f:
        json.dump(data, f, sort_keys=True, separators=(',', ':'))
      cmd = [
        os.path.join(CLIENT_DIR, 'isolateserver.py'), 'archive',
        '-I', isolate, '--namespace', 'default-gzip', isolated,
      ]
      out = subprocess.check_output(cmd)
    return out.split(' ', 1)[0]
  finally:
    file_path.rmtree(tmp)