def test_extract_dev(self):
     st = create_setting()
     st.validate = False
     # create tmp dirs for extraction output
     st.inst_dir = tempfile.mkdtemp()
     st.true_dir = tempfile.mkdtemp()
     
     extract(st)
     
     # check no of files
     self.assertEqual(len(st.dev_true_fns), 
                      len(st.dev_part_fns))
     self.assertEqual(len(st.dev_inst_fns), 
                      len(st.dev_part_fns))
     
     # test loading a corpus file
     corpus = ParallelGraphCorpus(inf=st.dev_true_fns[0])
     
     # test loading a instances file
     inst = CorpusInst()
     inst.loadtxt(st.dev_inst_fns[0],
                  st.descriptor.dtype)
     self.assertEqual(len(corpus), len(inst))
     
     clean_inst(st)
     clean_true(st)

    def setUp(self):
        super(ImportTestCase, self).setUp()
        self.url = reverse_course_url('import_handler', self.course.id)
        self.content_dir = path(tempfile.mkdtemp())

        def touch(name):
            """ Equivalent to shell's 'touch'"""
            with file(name, 'a'):
                os.utime(name, None)

        # Create tar test files -----------------------------------------------
        # OK course:
        good_dir = tempfile.mkdtemp(dir=self.content_dir)
        os.makedirs(os.path.join(good_dir, "course"))
        with open(os.path.join(good_dir, "course.xml"), "w+") as f:
            f.write('<course url_name="2013_Spring" org="EDx" course="0.00x"/>')

        with open(os.path.join(good_dir, "course", "2013_Spring.xml"), "w+") as f:
            f.write('<course></course>')

        self.good_tar = os.path.join(self.content_dir, "good.tar.gz")
        with tarfile.open(self.good_tar, "w:gz") as gtar:
            gtar.add(good_dir)

        # Bad course (no 'course.xml' file):
        bad_dir = tempfile.mkdtemp(dir=self.content_dir)
        touch(os.path.join(bad_dir, "bad.xml"))
        self.bad_tar = os.path.join(self.content_dir, "bad.tar.gz")
        with tarfile.open(self.bad_tar, "w:gz") as btar:
            btar.add(bad_dir)

        self.unsafe_common_dir = path(tempfile.mkdtemp(dir=self.content_dir))

    def setUp(self):

        self.host_log_dir = tempfile.mkdtemp(prefix='host_log_dir.')
        self.volume = tempfile.mkdtemp(prefix='volume.')
        for logf in ['test1.log', 'test2.log']:
            with open(os.path.join(self.volume, logf), 'w') as logp:
                logp.write(logf)

def test_load_fail():
    # 1. test bad file path
    # 2. test non-json file
    # 3. test bad extensions
    # 4. test bad codecs

    def __test(filename, fmt):
        jams.load(filename, fmt=fmt)

    # Make a non-existent file
    tdir = tempfile.mkdtemp()
    yield raises(IOError)(__test), os.path.join(tdir, 'nonexistent.jams'), 'jams'
    os.rmdir(tdir)

    # Make a non-json file
    tdir = tempfile.mkdtemp()
    badfile = os.path.join(tdir, 'nonexistent.jams')
    with open(badfile, mode='w') as fp:
        fp.write('some garbage')
    yield raises(ValueError)(__test), os.path.join(tdir, 'nonexistent.jams'), 'jams'
    os.unlink(badfile)
    os.rmdir(tdir)

    tdir = tempfile.mkdtemp()
    for ext in ['txt', '']:
        badfile = os.path.join(tdir, 'nonexistent')
        yield raises(jams.ParameterError)(__test), '{:s}.{:s}'.format(badfile, ext), 'auto'
        yield raises(jams.ParameterError)(__test), '{:s}.{:s}'.format(badfile, ext), ext
        yield raises(jams.ParameterError)(__test), '{:s}.jams'.format(badfile), ext
    os.rmdir(tdir)

    def setUp(self):
        # Setup workspace
        tmpdir1 = os.path.realpath(tempfile.mkdtemp())
        tmpdir2 = os.path.realpath(tempfile.mkdtemp())
        os.makedirs(os.path.join(tmpdir1, 'swift'))

        self.workspace = Workspace(source_root=tmpdir1,
                                   build_root=tmpdir2)

        # Setup toolchain
        self.toolchain = host_toolchain()
        self.toolchain.cc = '/path/to/cc'
        self.toolchain.cxx = '/path/to/cxx'

        # Setup args
        self.args = argparse.Namespace(
            enable_tsan_runtime=False,
            compiler_vendor='none',
            swift_compiler_version=None,
            clang_compiler_version=None,
            swift_user_visible_version=None,
            darwin_deployment_version_osx="10.9",
            benchmark=False,
            benchmark_num_onone_iterations=3,
            benchmark_num_o_iterations=3,
            enable_sil_ownership=False)

        # Setup shell
        shell.dry_run = True
        self._orig_stdout = sys.stdout
        self._orig_stderr = sys.stderr
        self.stdout = StringIO()
        self.stderr = StringIO()
        sys.stdout = self.stdout
        sys.stderr = self.stderr

    def test_ensure_tree(self):

        # Create source tree.

        # Create a pair of tempfile.
        tempdir_1, tempdir_2 = tempfile.mkdtemp(), tempfile.mkdtemp()

        with open(tempdir_1 + '/file1', 'w') as fd:
            fd.write('content1')

        os.mkdir(tempdir_1 + '/dir')
        with open(tempdir_1 + '/dir/file2', 'w') as fd:
            fd.write('content2')

        from espresso.helpers import fs

        # Ensure tree.
        fs.ensure_tree(tempdir_2, tempdir_1)

        # Assert for existence of paths.
        self.assertTrue(os.path.exists(tempdir_2 + '/dir'))
        self.assertTrue(os.path.exists(tempdir_2 + '/file1'))
        self.assertTrue(os.path.exists(tempdir_2 + '/dir/file2'))

        # Asssert files and dirs
        self.assertTrue(os.path.isdir(tempdir_2 + '/dir'))
        self.assertTrue(os.path.isfile(tempdir_2 + '/file1'))
        self.assertTrue(os.path.isfile(tempdir_2 + '/dir/file2'))

        # Assert for content in files.
        with open(tempdir_2 + '/file1') as fd:
            self.assertEqual(fd.read(), 'content1')

        with open(tempdir_2 + '/dir/file2') as fd:
            self.assertEqual(fd.read(), 'content2')

    def test_prep_sffs_in_dir_no_trim(self):
        """test_prep_sffs_in_dir should use the no_trim option only if sffinfo exists."""
        output_dir = tempfile.mkdtemp()
        gz_output_dir = tempfile.mkdtemp()

        try:
            check_sffinfo()
            perform_test = True
        except:
            perform_test = False

        if perform_test:
            prep_sffs_in_dir(self.sff_dir, output_dir, make_flowgram=False,
                             convert_to_flx=False, use_sfftools=True,
                             no_trim=True)

            fna_fp = os.path.join(output_dir, 'test.fna')

            self.assertEqual(open(fna_fp).read(), fna_notrim_txt)

            qual_fp = os.path.join(output_dir, 'test.qual')
            self.assertEqual(open(qual_fp).read(), qual_notrim_txt)

            self.assertRaises(TypeError, "gzipped SFF", prep_sffs_in_dir,
                              self.gz_sff_dir, gz_output_dir, make_flowgram=False,
                              convert_to_flx=False, use_sfftools=True,
                              no_trim=True)

            shutil.rmtree(output_dir)
            shutil.rmtree(gz_output_dir)

 def _runEvaluator(self, predFilePath, goldPath):
     tempDir = tempfile.mkdtemp()
     evaluatorDir = os.path.join(Settings.DATAPATH, "tools", "evaluators", "ChemProtEvaluator")
     removeTemp = False
     if tempDir == None:
         tempDir = tempfile.mkdtemp()
         removeTemp = True
     print >> sys.stderr, "Using temporary evaluation directory", tempDir
     evaluatorTempDir = os.path.join(tempDir, "ChemProtEvaluator")
     shutil.copytree(evaluatorDir, evaluatorTempDir)
     currentDir = os.getcwd()
     os.chdir(evaluatorTempDir)
     command = "java -cp bc6chemprot_eval.jar org.biocreative.tasks.chemprot.main.Main " + os.path.abspath(predFilePath) + " " + goldPath
     print >> sys.stderr, "Running CP17 evaluator: " + command
     p = subprocess.Popen(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
     for s in ["".join(x.readlines()).strip() for x in (p.stderr, p.stdout)]:
         if s != "":
             print >> sys.stderr, s
     os.chdir(currentDir)
     results = {}
     with open(os.path.join(evaluatorTempDir, "out", "eval.txt"), "rt") as f:
         for line in f:
             if ":" in line:
                 print >> sys.stderr, line.strip()
                 key, value = [x.strip() for x in line.split(":")]
                 value = float(value) if ("." in value or value == "NaN") else int(value)
                 assert key not in results
                 results[key] = value
     if removeTemp:
         print >> sys.stderr, "Removing temporary evaluation directory", tempDir
         shutil.rmtree(tempDir)
     return results

    def setUp(self):
        # Setup workspace
        tmpdir1 = os.path.realpath(tempfile.mkdtemp())
        tmpdir2 = os.path.realpath(tempfile.mkdtemp())
        os.makedirs(os.path.join(tmpdir1, "llvm"))

        self.workspace = Workspace(source_root=tmpdir1, build_root=tmpdir2)

        # Setup toolchain
        self.toolchain = host_toolchain()
        self.toolchain.cc = "/path/to/cc"
        self.toolchain.cxx = "/path/to/cxx"

        # Setup args
        self.args = argparse.Namespace(
            llvm_targets_to_build="X86;ARM;AArch64;PowerPC;SystemZ",
            llvm_assertions="true",
            compiler_vendor="none",
            clang_compiler_version=None,
            clang_user_visible_version=None,
            darwin_deployment_version_osx="10.9",
        )

        # Setup shell
        shell.dry_run = True
        self._orig_stdout = sys.stdout
        self._orig_stderr = sys.stderr
        self.stdout = StringIO()
        self.stderr = StringIO()
        sys.stdout = self.stdout
        sys.stderr = self.stderr

    def test_unarchive(self):
        import zipfile, tarfile

        good_archives = (
            ('good.zip', zipfile.ZipFile, 'r', 'namelist'),
            ('good.tar', tarfile.open, 'r', 'getnames'),
            ('good.tar.gz', tarfile.open, 'r:gz', 'getnames'),
            ('good.tar.bz2', tarfile.open, 'r:bz2', 'getnames'),
        )
        bad_archives = ('bad.zip', 'bad.tar', 'bad.tar.gz', 'bad.tar.bz2')

        for name, cls, mode, lister in good_archives:
            td = tempfile.mkdtemp()
            archive = None
            try:
                name = os.path.join(HERE, name)
                unarchive(name, td)
                archive = cls(name, mode)
                names = getattr(archive, lister)()
                for name in names:
                    p = os.path.join(td, name)
                    self.assertTrue(os.path.exists(p))
            finally:
                shutil.rmtree(td)
                if archive:
                    archive.close()

        for name in bad_archives:
            name = os.path.join(HERE, name)
            td = tempfile.mkdtemp()
            try:
                self.assertRaises(ValueError, unarchive, name, td)
            finally:
                shutil.rmtree(td)

def rm_rf(path, verbose=False):
    if not on_win and islink(path):
        # Note that we have to check if the destination is a link because
        # exists('/path/to/dead-link') will return False, although
        # islink('/path/to/dead-link') is True.
        if verbose:
            print "Removing: %r (link)" % path
        os.unlink(path)

    elif isfile(path):
        if verbose:
            print "Removing: %r (file)" % path
        if on_win:
            try:
                os.unlink(path)
            except (WindowsError, IOError):
                os.rename(path, join(tempfile.mkdtemp(), basename(path)))
        else:
            os.unlink(path)

    elif isdir(path):
        if verbose:
            print "Removing: %r (directory)" % path
        if on_win:
            try:
                shutil.rmtree(path)
            except (WindowsError, IOError):
                os.rename(path, join(tempfile.mkdtemp(), basename(path)))
        else:
            shutil.rmtree(path)

def test_switch_cache():
    """Test changing cache directory while extension is active."""
    global _counter

    dir1 = tempfile.mkdtemp(prefix='mdp-tmp-joblib-cache.',
                            dir=py.test.mdp_tempdirname)
    dir2 = tempfile.mkdtemp(prefix='mdp-tmp-joblib-cache.',
                            dir=py.test.mdp_tempdirname)
    x = mdp.numx.array([[10]], dtype='d')

    mdp.caching.activate_caching(cachedir=dir1)

    node = _CounterNode()
    _counter = 0
    node.execute(x)
    assert _counter == 1
    node.execute(x)
    assert _counter == 1

    # now change path
    mdp.caching.set_cachedir(cachedir=dir2)
    node.execute(x)
    assert _counter == 2
    node.execute(x)
    assert _counter == 2

    mdp.caching.deactivate_caching()

    def begin(self):
        #
        # We monkey-patch javabridge.start_vm here in order to
        # set up the ImageJ event bus (actually 
        # org.bushe.swing.event.ThreadSafeEventService) to not start
        # its cleanup thread which semi-buggy hangs around forever
        # and prevents Java from exiting.
        #
        def patch_start_vm(*args, **kwargs):
            result = start_vm(*args, **kwargs)
            if javabridge.get_env() is not None:
                try:
                    event_service_cls = javabridge.JClassWrapper(
                        "org.bushe.swing.event.ThreadSafeEventService")
                    event_service_cls.CLEANUP_PERIOD_MS_DEFAULT = None
                except:
                    pass
            return result
        patch_start_vm.func_globals["start_vm"] = javabridge.start_vm
        javabridge.start_vm = patch_start_vm
        if "CP_EXAMPLEIMAGES" in os.environ:
            self.temp_exampleimages = None
        else:
            self.temp_exampleimages = tempfile.mkdtemp(prefix="cpexampleimages")

        if "CP_TEMPIMAGES" in os.environ:
            self.temp_images = None
        else:
            self.temp_images = tempfile.mkdtemp(prefix="cptempimages")

 def test_extract_val_binary(self):
     st = create_setting()
     st.develop = False
     # create tmp dirs for extraction output
     st.inst_dir = tempfile.mkdtemp()
     st.true_dir = tempfile.mkdtemp()
     st.binary = True
     
     extract(st)
     
     # check no of files
     self.assertEqual(len(st.val_true_fns), 
                      len(st.val_part_fns))
     self.assertEqual(len(st.val_inst_fns), 
                      len(st.val_part_fns))
     
     # test loading a corpus file
     corpus = ParallelGraphCorpus(inf=st.val_true_fns[0])
     
     # test loading a instances file
     inst = CorpusInst()
     inst.loadbin(st.val_inst_fns[0])
     self.assertEqual(len(corpus), len(inst))
     
     clean_inst(st)
     clean_true(st)

 def setUp(self):
     self.temp_dir = tempfile.mkdtemp()
     self.storage = self.storage_class(location=self.temp_dir,
         base_url='/test_media_url/')
     # Set up a second temporary directory which is ensured to have a mixed
     # case name.
     self.temp_dir2 = tempfile.mkdtemp(suffix='aBc')

    def test_transform_function_serializer_failure(self):
        inputd = tempfile.mkdtemp()
        self.cpd = tempfile.mkdtemp("test_transform_function_serializer_failure")

        def setup():
            conf = SparkConf().set("spark.default.parallelism", 1)
            sc = SparkContext(conf=conf)
            ssc = SnappyStreamingContext(sc, 0.5)

            # A function that cannot be serialized
            def process(time, rdd):
                sc.parallelize(range(1, 10))

            ssc.textFileStream(inputd).foreachRDD(process)
            return ssc

        self.ssc = SnappyStreamingContext.getOrCreate(self.cpd, setup)
        try:
            self.ssc.start()
        except:
            import traceback
            failure = traceback.format_exc()
            self.assertTrue(
                    "It appears that you are attempting to reference SparkContext" in failure)
            return

        self.fail("using SparkContext in process should fail because it's not Serializable")

    def testSendOnlyToLitleSFTP_WithPreviouslyCreatedFile(self):
        requestFileName = "litleSdk-testBatchFile-testSendOnlyToLitleSFTP_WithPreviouslyCreatedFile.xml"
        request = litleBatchFileRequest(requestFileName)
        requestFile = request.requestFile.name
        self.assertTrue(os.path.exists(requestFile))
        configFromFile = request.config
        self.assertEqual('prelive.litle.com', configFromFile.batchHost)
        self.assertEqual('15000', configFromFile.batchPort)
        requestDir = configFromFile.batchRequestFolder
        responseDir = configFromFile.batchResponseFolder
        self.prepareTestRequest(request)
        request.prepareForDelivery()
        self.assertTrue(os.path.exists(requestFile))
        self.assertTrue(os.path.getsize(requestFile) > 0)

        tempfile.mkdtemp()
        newRequestDir = tempfile.gettempdir() + '/' + 'request'
        if not os.path.exists(newRequestDir):
            os.makedirs(newRequestDir)
        newRequestFileName = 'litle.xml'
        shutil.copyfile(requestFile, newRequestDir + '/' + newRequestFileName)
        configForRequest2 = copy.deepcopy(configFromFile)
        configForRequest2.batchRequestFolder = newRequestDir

        request2 = litleBatchFileRequest(newRequestFileName, configForRequest2)
        request2.sendRequestOnlyToSFTP(True)

        request3 = litleBatchFileRequest(newRequestFileName, configForRequest2)
        response = request3.retrieveOnlyFromSFTP()

        self.assertPythonApi(request3, response)

        self.assertGeneratedFiles(newRequestDir, responseDir, newRequestFileName, request3)

    def test_subreadset_consolidate(self):
        log.debug("Test methods directly")
        aln = SubreadSet(data.getXml(10), data.getXml(13))
        self.assertEqual(len(aln.toExternalFiles()), 2)
        outdir = tempfile.mkdtemp(suffix="dataset-unittest")
        outfn = os.path.join(outdir, 'merged.bam')
        consolidateBams(aln.toExternalFiles(), outfn, filterDset=aln)
        self.assertTrue(os.path.exists(outfn))
        consAln = SubreadSet(outfn)
        self.assertEqual(len(consAln.toExternalFiles()), 1)
        for read1, read2 in zip(sorted(list(aln)), sorted(list(consAln))):
            self.assertEqual(read1, read2)
        self.assertEqual(len(aln), len(consAln))

        log.debug("Test through API")
        aln = SubreadSet(data.getXml(10), data.getXml(13))
        self.assertEqual(len(aln.toExternalFiles()), 2)
        outdir = tempfile.mkdtemp(suffix="dataset-unittest")
        outfn = os.path.join(outdir, 'merged.bam')
        aln.consolidate(outfn)
        self.assertTrue(os.path.exists(outfn))
        self.assertEqual(len(aln.toExternalFiles()), 1)
        nonCons = SubreadSet(data.getXml(10), data.getXml(13))
        self.assertEqual(len(nonCons.toExternalFiles()), 2)
        for read1, read2 in zip(sorted(list(aln)), sorted(list(nonCons))):
            self.assertEqual(read1, read2)
        self.assertEqual(len(aln), len(nonCons))

    def test_prep_sffs_in_dir_FLX(self):
        """test_prep_sffs_in_dir should convert to FLX read lengths."""
        output_dir = tempfile.mkdtemp()
        gz_output_dir = tempfile.mkdtemp()

        prep_sffs_in_dir(
            self.sff_dir, output_dir, make_flowgram=True, convert_to_flx=True)
        prep_sffs_in_dir(
            self.gz_sff_dir, gz_output_dir, make_flowgram=True, convert_to_flx=True)

        fna_fp = os.path.join(output_dir, 'test_FLX.fna')
        fna_gz_fp = os.path.join(gz_output_dir, 'test_gz_FLX.fna')
        self.assertEqual(open(fna_fp).read(), fna_txt)
        self.assertEqual(open(fna_gz_fp).read(), fna_txt)

        qual_fp = os.path.join(output_dir, 'test_FLX.qual')
        qual_gz_fp = os.path.join(gz_output_dir, 'test_gz_FLX.qual')
        self.assertEqual(open(qual_fp).read(), qual_txt)
        self.assertEqual(open(qual_gz_fp).read(), qual_txt)

        flow_fp = os.path.join(output_dir, 'test_FLX.txt')
        flow_gz_fp = os.path.join(gz_output_dir, 'test_gz_FLX.txt')
        self.assertEqual(open(flow_fp).read(), flx_flow_txt)
        self.assertEqual(open(flow_gz_fp).read(), flx_flow_txt)

        shutil.rmtree(output_dir)
        shutil.rmtree(gz_output_dir)

def create_temp_dir(prefix = None):
    """ Create temporary directory with optional prefix.
    """
    if prefix is None:
        return tempfile.mkdtemp(prefix="{}-build.".format(PACKAGE_NAME))
    else:
        return tempfile.mkdtemp(prefix=prefix)