# Path: crumbs/utils/test_utils.py
# TEST_DATA_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), '..',
#                                              '..', 'test', 'test_data'))
# 
# Path: crumbs/utils/file_utils.py
# def compress_with_bgzip(in_fhand, compressed_fhand):
#     '''It compresses the input fhand.
# 
#     The input fhand has to be a real file with a name.
#     '''
#     cmd = ['bgzip', '-c', in_fhand.name]
#     check_call(cmd, stdout=compressed_fhand)
# 
# def uncompress_gzip(in_fhand, uncompressed_fhand):
#     '''It compresses the input fhand.
# 
#     The input fhand has to be a real file with a name.
#     '''
#     cmd = ['gzip', '-dc', in_fhand.name]
#     check_call(cmd, stdout=uncompressed_fhand)
# 
# def fhand_is_seekable(fhand):
#     'It returns True if the fhand is seekable'
#     try:
#         try:
#             # The stdin stream in some instances has seek, has no seekable
#             fhand.tell()
#         except IOError:
#             return False
#         try:
#             if fhand.seekable():
#                 return True
#             else:
#                 return False
#         except AttributeError:
#             return True
#     except AttributeError:
#         return False
# 
# 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 index_vcf_with_tabix(in_fpath):
#     cmd = ['tabix', '-p', 'vcf', in_fpath]
#     tabix = Popen(cmd, stdout=PIPE, stderr=PIPE)
#     stdout, stderr = tabix.communicate()
#     if tabix.returncode:
#         msg = 'Some problem indexing with tabix.\n'
#         msg += 'stdout: ' + stdout
#         msg += 'stderr: ' + stderr
#         raise RuntimeError(msg)


import unittest
import gzip
from tempfile import NamedTemporaryFile
from os.path import exists
from os.path import join as pjoin
from os import remove
from StringIO import StringIO
from crumbs.utils.test_utils import TEST_DATA_DIR
from crumbs.utils.file_utils import (compress_with_bgzip, uncompress_gzip,
                                     fhand_is_seekable,
                                     wrap_in_buffered_reader,
                                     index_vcf_with_tabix)
        # a wrapped BufferedReader
        fhand2 = wrap_in_buffered_reader(fhand)
        assert fhand_is_seekable(fhand2)

        # pylint: disable=R0903
        # pylint: disable=C0111
        class NonSeekable(object):
            'Just for testing'
            pass

        assert not fhand_is_seekable(NonSeekable())

        class NonSeekable2(object):
            def seek(self):
                pass

            def seekable(self):
                return False

        assert not fhand_is_seekable(NonSeekable2())


class CompressTest(unittest.TestCase):
    def test_bgzip_compression(self):
        orig = 'hola\ncaracola\n'
        orig_fhand = NamedTemporaryFile()
        orig_fhand.write(orig)
        orig_fhand.flush()

        compressed_fhand = NamedTemporaryFile(suffix='.gz')

# Path: crumbs/utils/test_utils.py
# TEST_DATA_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), '..',
#                                              '..', 'test', 'test_data'))
# 
# Path: crumbs/utils/file_utils.py
# def compress_with_bgzip(in_fhand, compressed_fhand):
#     '''It compresses the input fhand.
# 
#     The input fhand has to be a real file with a name.
#     '''
#     cmd = ['bgzip', '-c', in_fhand.name]
#     check_call(cmd, stdout=compressed_fhand)
# 
# def uncompress_gzip(in_fhand, uncompressed_fhand):
#     '''It compresses the input fhand.
# 
#     The input fhand has to be a real file with a name.
#     '''
#     cmd = ['gzip', '-dc', in_fhand.name]
#     check_call(cmd, stdout=uncompressed_fhand)
# 
# def fhand_is_seekable(fhand):
#     'It returns True if the fhand is seekable'
#     try:
#         try:
#             # The stdin stream in some instances has seek, has no seekable
#             fhand.tell()
#         except IOError:
#             return False
#         try:
#             if fhand.seekable():
#                 return True
#             else:
#                 return False
#         except AttributeError:
#             return True
#     except AttributeError:
#         return False
# 
# 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 index_vcf_with_tabix(in_fpath):
#     cmd = ['tabix', '-p', 'vcf', in_fpath]
#     tabix = Popen(cmd, stdout=PIPE, stderr=PIPE)
#     stdout, stderr = tabix.communicate()
#     if tabix.returncode:
#         msg = 'Some problem indexing with tabix.\n'
#         msg += 'stdout: ' + stdout
#         msg += 'stderr: ' + stderr
#         raise RuntimeError(msg)


# Path: test/test_utils.py
import unittest
import gzip
from tempfile import NamedTemporaryFile
from os.path import exists
from os.path import join as pjoin
from os import remove
from StringIO import StringIO
from crumbs.utils.test_utils import TEST_DATA_DIR
from crumbs.utils.file_utils import (compress_with_bgzip, uncompress_gzip,
                                     fhand_is_seekable,
                                     wrap_in_buffered_reader,
                                     index_vcf_with_tabix)

        # a wrapped BufferedReader
        fhand2 = wrap_in_buffered_reader(fhand)
        assert fhand_is_seekable(fhand2)

        # pylint: disable=R0903
        # pylint: disable=C0111
        class NonSeekable(object):
            'Just for testing'
            pass

        assert not fhand_is_seekable(NonSeekable())

        class NonSeekable2(object):
            def seek(self):
                pass

            def seekable(self):
                return False

        assert not fhand_is_seekable(NonSeekable2())


class CompressTest(unittest.TestCase):
    def test_bgzip_compression(self):
        orig = 'hola\ncaracola\n'
        orig_fhand = NamedTemporaryFile()
        orig_fhand.write(orig)
        orig_fhand.flush()

        compressed_fhand = NamedTemporaryFile(suffix='.gz')

        compress_with_bgzip(orig_fhand, compressed_fhand)