class TemporaryDir(object):
    '''This class creates temporary directories '''
    def __init__(self, suffix='', prefix='', directory=None):
        '''It initiates the class.'''
        self._name = tempfile.mkdtemp(suffix=suffix, prefix=prefix,
                                      dir=directory)

    @property
    def name(self):
        'Returns path to the dict'
        return self._name

    def close(self):
        '''It removes the temp dir'''
        if os.path.exists(self._name):
            shutil.rmtree(self._name) def rel_symlink(path1, path2):
    'It makes the relative to path1 from within dirname(path2)'
    # The working directory can not be used because it is not thread safe
    # working directory is a global environment variable shared by all threads
    path1 = os.path.abspath(path1)
    path2 = os.path.abspath(path2)
    rel_path1 = _rel_path(path1, path2)

    fname2 = os.path.split(path2)[-1]

    temp_dir = tempfile.mkdtemp()
    try:
        temp_link_path = os.path.join(temp_dir, fname2)

        os.symlink(rel_path1, temp_link_path)
        shutil.move(temp_link_path, path2)
    finally:
        os.rmdir(temp_dir) def wrap_in_buffered_reader(fhand, force_wrap=False,
                            buffering=DEF_FILE_BUFFER):
    '''It wraps the given file in a peekable BufferedReader.

    If the file is seekable it doesn't do anything.
    '''
    if not force_wrap and fhand_is_seekable(fhand):
        return fhand
    else:
        fhand = io.open(fhand.fileno(), mode='rb',
                        buffering=buffering)  # with text there's no peek

    return fhand def check_process_finishes(process, binary, stdout=None, stderr=None):
    'It checks that the given process finishes OK, otherwise raises an Error'

    stdout_msg, stderr_msg = process.communicate()

    returncode = process.returncode
    if returncode == 0:
        return
    elif returncode is None:
        msg = 'The process for {:s} is still running with PID'
        msg = msg.format(binary, process.PID)
        raise RuntimeError(msg)

    if stdout and not stdout_msg:
        stdout.flush()
        stdout_msg = open(stdout.name).read()
    if stderr and not stderr_msg:
        stderr.flush()
        stderr_msg = open(stderr.name).read()
    msg = '{:s} had a problem running\n'.format(binary)
    if stdout_msg:
        msg += 'stdout:\n{:s}\n'.format(stdout_msg)
    if stderr_msg:
        msg += 'stderr:\n{:s}\n'.format(stderr_msg)
    raise ExternalBinaryError(msg) def popen(*args, **kwargs):
    'It spawns a process with subprocess.Popen'
    try:
        return Popen(*args, **kwargs)
    except OSError, error:
        msg = 'The binary "%s" is not in the path.' % args[0][0]
        if 'No such file' in str(error):
            raise MissingBinaryError(msg) BIN_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..',
                                       'bin')) class ExternalBinaryError(Exception):
    'A call to an external binary failed'
    pass class MissingBinaryError(Exception):
    'A call to an external binary failed becase the binary is not in the path'
    pass TEST_DATA_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), '..',
                                             '..', 'test', 'test_data')) ERROR_ENVIRON_VARIABLE = 'seq_crumbs_error_debugging' def guess_seq_type(fhand):
    '''It guesses if the file is nucleotide or protein'''
    rna = set(ambiguous_rna_letters)
    dna = set(ambiguous_dna_letters)
    rna_dna = rna.union(dna)

    protein = set(extended_protein_letters)
    only_prot = list(protein.difference(rna_dna))

    chunk_size = 1024
    chunk = peek_chunk_from_file(fhand, chunk_size)
    if not chunk:
        raise UnknownFormatError('The file is empty')
    fhand_ = cStringIO.StringIO(chunk)
    total_letters = 0
    nucleotides = 0
    for seq in read_seqs([fhand_]):
        for letter in get_str_seq(seq):
            total_letters += 1
            if letter in ('gcatnuGCATNU'):
                nucleotides += 1
            if letter in only_prot:
                return 'prot'
    nucl_freq = nucleotides / total_letters
    if nucl_freq > 0.8:
        return 'nucl'

    raise RuntimeError('unable to guess the seq type') def get_setting(key):
    'It returns the value for one setting'
    return _settings[key] def get_format(fhand):
    'It gets the format or it looks in the inventory'
    id_ = _get_fhand_id(fhand)
    try:
        file_format = FILEFORMAT_INVENTORY[id_]
    except KeyError:
        file_format = None

    if file_format is None:
        file_format = _guess_format(fhand, force_file_as_non_seek=False)
        FILEFORMAT_INVENTORY[id_] = file_format

    return file_format FILEFORMAT_INVENTORY = {} def set_format(fhand, file_format):
    'It sets the file format in the global inventory variable'
    id_ = _get_fhand_id(fhand)
    if id_ in FILEFORMAT_INVENTORY:
        msg = 'The given instance already setted its file format'
        raise RuntimeError(msg)
    FILEFORMAT_INVENTORY[id_] = file_format class SqliteCache(object):
    def __init__(self, fpath):
        self._cache_table_name = 'cachedata'
        self._fhand = None
        if not os.path.exists(fpath):
            self._fhand = open(fpath, 'w')
            fpath = self._fhand.name

        self.connection = sqlite3.connect(fpath)
        self._create_init_table()

    def __enter__(self):
        return self

    def __exit__(self, type, value, traceback):
        self.close()
        if type:
            return 0

    def _create_init_table(self):
        c = self.connection.cursor()
        c.execute("SELECT name FROM sqlite_master WHERE type='table';")
        result = c.fetchall()
        if not result or self._cache_table_name not in result[0]:
            sql = 'create table {table_name} (key TEXT, value BLOB)'
            c.execute(sql.format(table_name=self._cache_table_name))
        self.connection.commit()

    def __getitem__(self, key):
        c = self.connection.cursor()
        sql = "select * from {cache_table} where key = '{key}'"
        c.execute(sql.format(key=key, cache_table=self._cache_table_name))

        result = c.fetchall()
        if not result:
            return None
        if len(result) > 1:
            msg = 'More than one ocurrence of the key:{}'.format(key)
            raise RuntimeError(msg)

        return pickle.loads(str(result[0][1]))

    def __setitem__(self, key, value):
        c = self.connection.cursor()
        value = pickle.dumps(value, pickle.HIGHEST_PROTOCOL)
        if key in self:
            sql = "update {cache_table} set value = ? where key = ?"
            sql = sql.format(cache_table=self._cache_table_name, key=key)
            c.execute(sql, [sqlite3.Binary(value), key])
        else:
            sql = "insert into {cache_table} values(?, ?)"
            sql = sql.format(cache_table=self._cache_table_name)
            c.execute(sql, [key, sqlite3.Binary(value)])
        self.connection.commit()

    def close(self):
        if self._fhand is not None:
            self._fhand.close()
        self.connection.close()

    def __contains__(self, value):
        c = self.connection.cursor()
        sql = "select key from {cache_table} where key='{key}'"
        c.execute(sql.format(cache_table=self._cache_table_name, key=value))
        return True if c.fetchone() else False

    def dump(self):
        c = self.connection.cursor()
        sql = "select * from {}".format(self._cache_table_name)
        for key, value in c.execute(sql):
            value = pickle.loads(str(value))
            yield key, value

    def __str__(self):
        c = self.connection.cursor()
        text = None
        sql = "select * from {}".format(self._cache_table_name)
        for key, value in c.execute(sql):
            value = pickle.loads(str(value))
            if text is None:
                key_order = value.keys()
                text = ['\t'.join(['key'] + key_order)]
            line = [key]
            for valkey in key_order:
                try:
                    v = str(value[valkey]) if value[valkey] else ''
                except KeyError:
                    v = ''
                line.append(v)
            text.append('\t'.join(line))
        return '\n'.join(text) + '\n'
import unittest
import os
import sys
import hashlib
from subprocess import Popen, PIPE, check_output, CalledProcessError
from tempfile import NamedTemporaryFile
from cStringIO import StringIO
from crumbs.utils.file_utils import (TemporaryDir, rel_symlink,
                                     wrap_in_buffered_reader)
from crumbs.utils.bin_utils import check_process_finishes, popen, BIN_DIR
from crumbs.exceptions import ExternalBinaryError, MissingBinaryError
from crumbs.utils.test_utils import TEST_DATA_DIR
from crumbs.utils.tags import ERROR_ENVIRON_VARIABLE
from crumbs.seq.seqio import guess_seq_type
from crumbs.settings import get_setting
from crumbs.seq.utils.file_formats import (get_format, FILEFORMAT_INVENTORY,
                                           set_format)
from crumbs.utils.sqlite_utils import SqliteCache
            assert binary in str(error)

        cmd = [binary]
        process = Popen(cmd, stdout=PIPE)
        check_process_finishes(process, binary)

    def test_popen(self):
        'It checks that we can create a process'

        try:
            popen(['bad_binary'])
            self.fail()
        except MissingBinaryError:
            pass

        popen(['ls'], stdout=PIPE)

    def test_io_open(self):
        'It checks the file wrapping in ReaderBuffers'

        # a standard file
        fhand = NamedTemporaryFile()
        fhand.write('hola')
        fhand.flush()
        fhand2 = open(fhand.name)
        fhand3 = wrap_in_buffered_reader(fhand2, force_wrap=True)
        assert fhand3.peek(10) == 'hola'

    @staticmethod
    def test_guess_seq_type():
        'It checks that we can guess the type of the seqs in a file'
        fpath = os.path.join(TEST_DATA_DIR, 'arabidopsis_genes')
        guess_seq_type(open(fpath))

    def test_get_format_fhand(self):
        "It checks the get/set format functions"
        #file fhand
        fhand = NamedTemporaryFile()
        fhand.write('>seq\natgctacgacta\n')
        fhand.flush()
        name = fhand.name
        id_ = id(fhand)

        file_format = get_format(fhand)
        assert FILEFORMAT_INVENTORY[(id_, name)] == file_format
        num_keys = len(FILEFORMAT_INVENTORY)

        file_format = get_format(fhand)
        assert FILEFORMAT_INVENTORY[(id_, name)] == file_format
        assert len(FILEFORMAT_INVENTORY) == num_keys

        fhand = NamedTemporaryFile()
        set_format(fhand, 'fasta')

        assert 'fasta' == get_format(fhand)

    def test_get_format_stringio(self):
        "It checks the get/set format functions"
        #stiongIO
        stringIO_fhand = StringIO('>seq\natgctacgacta\n')

        striongIOhash = hashlib.sha224(stringIO_fhand.getvalue()[:100]).hexdigest()
        id_ = id(stringIO_fhand)

        file_format = get_format(stringIO_fhand)
        assert FILEFORMAT_INVENTORY[(id_, striongIOhash)] == file_format


class ErrorHandlingTest(unittest.TestCase):
    'It tests the handling of the unexpected errors'
    def test_error_handling(self):
        'We handle the errors'
        cat_bin = os.path.join(BIN_DIR, 'cat_seqs')

        os.environ[ERROR_ENVIRON_VARIABLE] = 'Fail'
        stderr = NamedTemporaryFile(suffix='.error')
        try:
            check_output([cat_bin, '-h'], stderr=stderr)
            self.fail('Error expected')
        except CalledProcessError:
            pass
        finally:
            del os.environ[ERROR_ENVIRON_VARIABLE]
            stderr.close()


class SettingsTest(unittest.TestCase):
    'It tests the get_settings function'
    def test_get_settings(self):
        'We get the settings'
        kmer_size = get_setting('DEFAULT_KMER_SIZE')
        assert kmer_size

    def test_environment_settings(self):
        'It sets settings with the environment variables'
        code = 'from crumbs.settings import get_setting\n'
        code += 'print get_setting("DEFAULT_KMER_SIZE")\n'

        test_script = NamedTemporaryFile()
        test_script.write(code)
        test_script.flush()

        environ = {key: val for key, val in os.environ.items()}
        environ['SEQ_CRUMBS_DEFAULT_KMER_SIZE'] = '1'
        cmd = [sys.executable, test_script.name]
        out = check_output(cmd, env=environ)
        assert out == '1\n'


class SqliteTest(unittest.TestCase):
    def test_init(self):
        fhand = NamedTemporaryFile()
        with SqliteCache(fhand.name) as sqlitecache:
            assert sqlitecache['seq1'] is None
            sqlitecache['seq1'] = 23
            assert sqlitecache['seq1'] == 23
            assert 'seq1' in sqlitecache
            assert 'seq2' not in sqlitecache

            # reopen the same cache file

            sqlitecache2 = SqliteCache(fhand.name)