How to use the twitter.common.contextutil.temporary_dir function in twitter

def test_task_health_failed(self):
    proxy_driver = ProxyDriver()
    with SignalServer(UnhealthyHandler) as port:
      with temporary_dir() as checkpoint_root:
        health_check_config = HealthCheckConfig(initial_interval_secs=0.1, interval_secs=0.1)
        _, executor = make_executor(proxy_driver,
                                    checkpoint_root,
                                    MESOS_JOB(task=SLEEP60, health_check_config=health_check_config),
                                    ports={'health': port},
                                    fast_status=True,
                                    status_providers=(HealthCheckerProvider(),))
        executor.terminated.wait()

    updates = proxy_driver.method_calls['sendStatusUpdate']
    assert len(updates) == 3
    assert updates[-1][0][0].state == mesos_pb.TASK_FAILED

def test_on_context_exit(self):
    with temporary_dir() as root_dir:
      parse_context = ParseContext(create_buildfile(root_dir, 'a'))
      with pytest.raises(parse_context.ContextError):
        parse_context.on_context_exit(lambda: 37)

    with temporary_dir() as root_dir:
      buildfile = create_buildfile(root_dir, 'a',
        content=dedent("""
          import os
          from pants.base.parse_context import ParseContext
          def leave_a_trail(file, contents=''):
            with open(file, 'w') as b:
              b.write(contents)
          b_file = os.path.join(os.path.dirname(__file__), 'b')
          ParseContext.locate().on_context_exit(leave_a_trail, b_file, contents='42')
          assert not os.path.exists(b_file), 'Expected context exit action to be delayed.'
        """).strip()
      )
      b_file = os.path.join(root_dir, 'a', 'b')
      self.assertFalse(os.path.exists(b_file))
      ParseContext(buildfile).parse()
      with open(b_file, 'r') as b:

def run_gc_with(active_executors, retained_tasks, lose=False):
  proxy_driver = ProxyDriver()
  with temporary_dir() as td:
    setup_tree(td, lose=lose)
    executor = ThinTestThermosGCExecutor(td, active_executors=active_executors)
    executor.registered(proxy_driver, None, None, None)
    executor.start()
    art = AdjustRetainedTasks(retainedTasks=retained_tasks)
    executor.launchTask(proxy_driver, serialize_art(art, TASK_ID))
    wait_until_not(lambda: executor._gc_task_queue, clock=executor._clock)
    wait_until_not(lambda: executor._task_id, clock=executor._clock)
    assert len(executor._gc_task_queue) == 0
    assert not executor._task_id
  assert len(proxy_driver.updates) >= 1
  if not lose: # if the task is lost it will be cleaned out of band (by clean_orphans),
               # so we don't care when the GC task actually finishes
    assert proxy_driver.updates[-1][0] == mesos.TASK_FINISHED
    assert proxy_driver.updates[-1][1] == TASK_ID
  return executor, proxy_driver

def yield_pex_builder(zip_safe=True, installer_impl=EggInstaller, interpreter=None):
  with nested(temporary_dir(),
              make_bdist('p1',
                         zipped=True,
                         zip_safe=zip_safe,
                         installer_impl=installer_impl,
                         interpreter=interpreter)) as (td, p1):
    pb = PEXBuilder(path=td, interpreter=interpreter)
    pb.add_dist_location(p1.location)
    yield pb

def add_products(self, context, product_type, target, *products):
    product_mapping = context.products.get(product_type)
    with temporary_dir() as outdir:
      def create_product(product):
        with safe_open(os.path.join(outdir, product), mode='w') as fp:
          fp.write(product)
        return product
      product_mapping.add(target, outdir, map(create_product, products))
      yield temporary_dir

def test_directory_sandbox():
  with temporary_dir() as d:
    ds1 = DirectorySandbox(os.path.join(d, 'task1'))
    ds2 = DirectorySandbox(os.path.join(d, 'task2'))
    ds1.create()
    ds2.create()
    assert os.path.exists(ds1.root)
    assert os.path.exists(ds2.root)
    ds1.destroy()
    assert not os.path.exists(ds1.root)
    assert os.path.exists(ds2.root)
    ds2.destroy()
    assert not os.path.exists(ds2.root)

def yield_runner(self, runner_class, **bindings):
    with contextlib.nested(temporary_dir(), temporary_dir()) as (td1, td2):
      sandbox = DirectorySandbox(td1)
      checkpoint_root = td2

      task_runner = runner_class(
          runner_pex=os.path.join('dist', 'thermos_runner.pex'),
          task_id='hello_world',
          task=TASK.bind(**bindings).task(),
          role=getpass.getuser(),
          portmap={},
          sandbox=sandbox,
          checkpoint_root=checkpoint_root,
      )

      yield task_runner

def run_gc_with_timeout(**kw):
  proxy_driver = ProxyDriver()
  with temporary_dir() as td:
    executor_class = make_gc_executor_with_timeouts(**kw)
    executor = executor_class(td)
    executor.registered(proxy_driver, None, None, None)
    executor.start()
    yield (proxy_driver, executor)

def test_cache_spec_parsing(self):
    artifact_root = '/bogus/artifact/root'

    def check(expected_type, spec):
      cache = create_artifact_cache(MockLogger(), artifact_root, spec, 'TestTask', 'testing')
      self.assertTrue(isinstance(cache, expected_type))
      self.assertEquals(cache.artifact_root, artifact_root)

    with temporary_dir() as tmpdir:
      cachedir = os.path.join(tmpdir, 'cachedir')  # Must be a real path, so we can safe_mkdir it.
      check(LocalArtifactCache, cachedir)
      check(RESTfulArtifactCache, 'http://localhost/bar')
      check(CombinedArtifactCache, [cachedir, 'http://localhost/bar'])