def test_stream():
    """
    Stream and file-based equality, both whole-file and Interval by
    Interval
    """
    a = pybedtools.example_bedtool('a.bed')
    b = pybedtools.example_bedtool('b.bed')
    c = a.intersect(b)

    # make an unwriteable dir...
    orig_tempdir = pybedtools.get_tempdir()
    if os.path.exists('unwriteable'):
        os.system('rm -rf unwriteable')
    os.system('mkdir unwriteable')
    os.system('chmod -w unwriteable')

    # ...set that to the new tempdir
    pybedtools.set_tempdir('unwriteable')

    # this should really not be written anywhere
    d = a.intersect(b, stream=True)

    assert_raises(NotImplementedError, c.__eq__, d)
    d_contents = d.fn.read()
    c_contents = open(c.fn).read()
    assert d_contents == c_contents

    # reconstruct d and check Interval-by-Interval equality
    pybedtools.set_tempdir('unwriteable')
    d = a.intersect(b, stream=True)

    for i,j in zip(c, d):
        assert str(i) == str(j)

    # Now do something similar with GFF files.
    a = pybedtools.example_bedtool('a.bed')
    f = pybedtools.example_bedtool('d.gff')

    # file-based
    pybedtools.set_tempdir(orig_tempdir)
    g1 = f.intersect(a)

    # streaming
    pybedtools.set_tempdir('unwriteable')
    g2 = f.intersect(a, stream=True)

    for i,j in zip(g1, g2):
        assert str(i) == str(j)

    # this was segfaulting at one point, just run to make sure
    g3 = f.intersect(a, stream=True)
    for i in iter(g3):
        print i

    for row in f.cut(range(3), stream=True):
        row[0], row[1], row[2]
        assert_raises(IndexError, row.__getitem__, 3)

    pybedtools.set_tempdir(orig_tempdir)
    os.system('rm -fr unwriteable')

def test_issue_141():
    a = pybedtools.example_bedtool('a.bed')
    b = pybedtools.example_bedtool('b.bed')

    # make an empty file
    empty = pybedtools.BedTool("", from_string=True)

    # invalid file format
    malformed = pybedtools.BedTool('a	a	a', from_string=True)

    # positive control; works
    a + b

    # "adding" an empty file always gets zero features
    assert len(a + empty) == 0
    assert len(empty + a) == 0
    assert len(empty + empty) == 0

    # "adding" a malformed file raises MalformedBedLineError
    # (an uncaught exception raised when trying to intersect)
    with pytest.raises(pybedtools.MalformedBedLineError):
        a + malformed

    x = pybedtools.example_bedtool('x.bam')
    x + a

def test_tuple_creation():
    # everything as a string
    t = [
            ("chr1", "1", "100", "feature1", "0", "+"),
            ("chr1", "100", "200", "feature2", "0", "+"),
            ("chr1", "150", "500", "feature3", "0", "-"),
            ("chr1", "900", "950", "feature4", "0", "+")
        ]
    x = pybedtools.BedTool(t).saveas()
    assert pybedtools.example_bedtool('a.bed') == x

    t = [
            ("chr1", 1, 100, "feature1", 0, "+"),
            ("chr1", 100, 200, "feature2", 0, "+"),
            ("chr1", 150, 500, "feature3", 0, "-"),
            ("chr1", 900, 950, "feature4", 0, "+")
        ]
    x = pybedtools.BedTool(t).saveas()
    assert pybedtools.example_bedtool('a.bed') == x

    t = [
            ("chr1", "fake", "gene", "50", "300", ".", "+", ".", "ID=gene1"),
            ("chr1", "fake", "mRNA", "50", "300", ".", "+", ".", "ID=mRNA1;Parent=gene1;"),
            ("chr1", "fake", "CDS", "75", "150", ".", "+", ".", "ID=CDS1;Parent=mRNA1;"),
            ("chr1", "fake", "CDS", "200", "275", ".", "+", ".", "ID=CDS2;Parent=mRNA1;"),
            ("chr1", "fake", "rRNA", "1200", "1275", ".", "+", ".", "ID=rRNA1;"),]
    x = pybedtools.BedTool(t).saveas()

    # Make sure that x has actual Intervals and not plain tuples or something
    assert isinstance(x[0], pybedtools.Interval)
    assert repr(x[0]) == "Interval(chr1:49-300)"
    assert x[0]['ID'] == 'gene1'

def test_cat():
    a = pybedtools.example_bedtool('a.bed')
    b = pybedtools.example_bedtool('b.bed')
    b_fn = pybedtools.example_filename('b.bed')
    assert a.cat(b) == a.cat(b_fn)
    expected =  fix("""
    chr1 1   500
    chr1 800 950
    """)
    assert a.cat(b) == expected

    a = pybedtools.example_bedtool('a.bed')
    b = pybedtools.example_bedtool('b.bed')
    c = a.cat(b, postmerge=False)
    assert len(a) + len(b) == len(c), (len(a), len(b), len(c))

    print c
    assert c == fix("""
    chr1	1	100	feature1	0	+
    chr1	100	200	feature2	0	+
    chr1	150	500	feature3	0	-
    chr1	900	950	feature4	0	+
    chr1	155	200	feature5	0	-
    chr1	800	901	feature6	0	+
    """)

def test_tail():
    a = pybedtools.example_bedtool('rmsk.hg18.chr21.small.bed')
    observed = a.tail(as_string=True)
    expected = fix(
        """
        chr21	13355834	13356047	MER58A	892	-
        chr21	13356250	13356290	AT_rich	26	+
        chr21	13356358	13356381	AT_rich	23	+
        chr21	13356571	13356910	L2	333	-
        chr21	13357179	13357987	L1MEc	1264	-
        chr21	13358003	13358300	L1MEc	379	-
        chr21	13358304	13358952	L1MEc	1271	-
        chr21	13358960	13359288	L2	336	+
        chr21	13359444	13359751	AluY	2337	+
        chr21	13360044	13360225	L1M5	284	-""")
    assert observed == expected


    # only ask for 3 lines
    observed = a.tail(3, as_string=True)
    expected = fix(
        """
        chr21	13358960	13359288	L2	336	+
        chr21	13359444	13359751	AluY	2337	+
        chr21	13360044	13360225	L1M5	284	-""")
    assert observed == expected


    # For short files, whole thing should be returned
    a = pybedtools.example_bedtool('a.bed')
    expected = str(a)
    obs = a.tail(as_string=True)
    assert obs == expected

def test_history_step():
    a = pybedtools.example_bedtool('a.bed')
    b = pybedtools.example_bedtool('b.bed')
    c = a.intersect(b)
    d = c.subtract(a)

    tag = c.history[0].result_tag
    assert pybedtools.find_tagged(tag) == c

    assert_raises(ValueError, pybedtools.find_tagged, 'nonexistent')


    print d.history
    d.delete_temporary_history(ask=True, raw_input_func=lambda x: 'n')
    assert os.path.exists(a.fn)
    assert os.path.exists(b.fn)
    assert os.path.exists(c.fn)
    assert os.path.exists(d.fn)

    d.delete_temporary_history(ask=True, raw_input_func=lambda x: 'Yes')
    assert os.path.exists(a.fn)
    assert os.path.exists(b.fn)
    assert not os.path.exists(c.fn) # this is the only thing that should change
    assert os.path.exists(d.fn)

    a = pybedtools.example_bedtool('a.bed')
    b = pybedtools.example_bedtool('b.bed')
    c = a.intersect(b)
    d = c.subtract(a)
    d.delete_temporary_history(ask=False)
    assert os.path.exists(a.fn)
    assert os.path.exists(b.fn)
    assert not os.path.exists(c.fn) # this is the only thing that should change
    assert os.path.exists(d.fn)

def test_bam_filetype():
    # regression test -- this was segfaulting before because IntervalFile
    # couldn't parse SAM
    a = pybedtools.example_bedtool('gdc.bam')
    b = pybedtools.example_bedtool('gdc.gff')
    c = a.intersect(b)
    assert c.file_type == 'bam'

def test_output_kwarg():
    a = pybedtools.example_bedtool("a.bed")
    b = pybedtools.example_bedtool("b.bed")
    c = a.intersect(b)
    d = a.intersect(b, output="deleteme.bed")
    assert c == d
    os.unlink("deleteme.bed")

def test_jaccard():
    x = pybedtools.example_bedtool('a.bed')

    results = x.jaccard(pybedtools.example_bedtool('b.bed'))
    assert results == {'intersection': 46, 'union': 649, 'jaccard': 0.0708783, 'n_intersections': 2}, results

    results2 = x.jaccard(pybedtools.example_bedtool('b.bed'), stream=True)
    assert results == results2, results2

def test_annotate_xstream():
    a = pybedtools.example_bedtool('m1.bed')
    b = pybedtools.example_bedtool('mm9.bed12')
    c = annotate.add_xstream(a, b, dist=1000, updown="up")
    assert a.field_count() == c.field_count() - 1
    assert len(a) == len(c)
    d = annotate.add_xstream(c, b, dist=1000, updown="down")
    assert a.field_count() == d.field_count() - 2

def test_random_intersection():
    # TODO:
    return
    N = 4
    a = pybedtools.example_bedtool('a.bed')
    b = pybedtools.example_bedtool('b.bed')
    li = list(a.randomintersection(b, N))
    assert len(li) == N, li

def run(d):
    method = d['method']
    bedtool = d['bedtool']
    convert = d['convert']
    kwargs = d['kw'].copy()
    expected = d['test_case']['expected']

    bedtool_converter = convert.pop('bedtool')
    bedtool = (
        converters[bedtool_converter](pybedtools.example_bedtool(bedtool))
    )

    for k, converter_name in convert.items():
        kwargs[k] = (
            converters[converter_name](pybedtools.example_bedtool(kwargs[k]))
        )
    result = getattr(bedtool, method)(**kwargs)
    res = str(result)
    expected = fix(expected)
    try:
        assert res == expected

    except AssertionError:
        print(result.fn)
        print('Method call:')
        args = []
        for key, val in list(kwargs.items()):
            args.append(('%s=%s' % (key, val)).strip())

        args = ', '.join(args)
        print('BedTool.%(method)s(%(args)s)' % locals())
        print('Got:')
        print(res)
        print('Expected:')
        print(expected)
        print('Diff:')
        for i in (
            difflib.unified_diff(res.splitlines(1), expected.splitlines(1))
        ):
            print(i, end=' ')

        # Make tabs and newlines visible
        spec_res = res.replace('\t', '\\t').replace('\n', '\\n\n')
        spec_expected = expected.replace('\t', '\\t').replace('\n', '\\n\n')

        print('Showing special characters:')
        print('Got:')
        print(spec_res)
        print('Expected:')
        print(spec_expected)
        print('Diff:')
        for i in (
            difflib.unified_diff(spec_res.splitlines(1),
                                 spec_expected.splitlines(1))
        ):
            print(i, end=' ')
        raise

def test_issue_147():
    # previously this would raise BEDToolsError because of unexpected stderr.
    with open(pybedtools.BedTool._tmp(), 'w') as tmp:
        orig_stderr = sys.stderr
        sys.stderr = tmp
        v = pybedtools.example_bedtool('vcf-stderr-test.vcf')
        b = pybedtools.example_bedtool('vcf-stderr-test.bed')
        v.intersect(b)
    sys.stderr = orig_stderr

def test_many_files():
    """regression test to make sure many files can be created
    """
    a = pybedtools.example_bedtool('a.bed')
    b = pybedtools.example_bedtool('b.bed')
    # Previously, IntervalFile would leak open files and would cause OSError
    # (too many open files) at iteration 1010 or so.
    for i in xrange(1100):
        c = a.intersect(b)

def test_annotate_closest():
    a = pybedtools.example_bedtool('m1.bed')
    b = pybedtools.example_bedtool('mm9.bed12')
    c = annotate.add_closest(a, b)
    assert len(a) == len(c), (len(a), len(c), str(c))
    assert a.field_count() == c.field_count() - 2
    # in this test-case, the final column should be exon;intron
    # since m1 completely contains both an exon and an intron.
    f = next(iter(c))

def test_chromsizes_in_5prime_3prime():

    # standard 5'
    a = pybedtools.example_bedtool('a.bed')\
        .each(featurefuncs.five_prime, 1, 10, add_to_name="_TSS",
              genome=pybedtools.chromsizes("hg19"))\
        .saveas()
    assert a == fix(
        """
        chr1	0	11	feature1_TSS	0	+
        chr1	99	110	feature2_TSS	0	+
        chr1	490	501	feature3_TSS	0	-
        chr1	899	910	feature4_TSS	0	+
        """), str(a)

    # add genomes sizes; last feature should be truncated
    a = pybedtools.example_bedtool('a.bed')\
        .each(featurefuncs.five_prime, 1, 10, add_to_name="_TSS",
              genome=dict(chr1=(0, 900)))\
        .saveas()
    assert a == fix(
        """
        chr1	0	11	feature1_TSS	0	+
        chr1	99	110	feature2_TSS	0	+
        chr1	490	501	feature3_TSS	0	-
        chr1	899	900	feature4_TSS	0	+
        """), str(a)

    # same thing but for 3'.
    # Note that the last feature chr1:949-960 is completely truncated because
    # it would entirely fall outside of the chromosome
    a = pybedtools.example_bedtool('a.bed')\
            .each(featurefuncs.three_prime, 1, 10, add_to_name="_TSS",
                 genome=dict(chr1=(0, 900)))\
            .saveas()
    assert a == fix(
        """
        chr1	99	110	feature1_TSS	0	+
        chr1	199	210	feature2_TSS	0	+
        chr1	140	151	feature3_TSS	0	-
        chr1	900	900	feature4_TSS	0	+
        """), str(a)

    # be a lot harsher with the chromsizes to ensure features on both strands
    # get truncated correctly
    a = pybedtools.example_bedtool('a.bed')\
            .each(featurefuncs.three_prime, 1, 10, add_to_name="_TSS",
                 genome=dict(chr1=(0, 120)))\
            .saveas()
    assert a == fix(
        """
        chr1	99	110	feature1_TSS	0	+
        chr1	120	120	feature2_TSS	0	+
        chr1	120	120	feature3_TSS	0	-
        chr1	120	120	feature4_TSS	0	+
        """), str(a)

def test_issue_118():
    p = psutil.Process(os.getpid())
    start_fds = p.num_fds()
    a = pybedtools.example_bedtool('a.bed')
    b = pybedtools.example_bedtool('b.bed')
    for i in range(100):
        c = a.intersect(b)
        c.field_count()
    stop_fds = p.num_fds()
    assert start_fds == stop_fds

def test_gzipped_files_can_be_intersected():
    agz = _make_temporary_gzip(pybedtools.example_filename('a.bed'))
    bgz = _make_temporary_gzip(pybedtools.example_filename('b.bed'))

    agz = pybedtools.BedTool(agz)
    bgz = pybedtools.BedTool(bgz)

    a = pybedtools.example_bedtool('a.bed')
    b = pybedtools.example_bedtool('b.bed')
    assert a.intersect(b) == agz.intersect(bgz) == a.intersect(bgz) == agz.intersect(b)

def test_repr_and_printing():
    a = pybedtools.example_bedtool('a.bed')
    b = pybedtools.example_bedtool('b.bed')
    c = a+b
    os.unlink(c.fn)
    assert 'a.bed' in repr(a)
    assert 'b.bed' in repr(b)
    assert 'MISSING FILE' in repr(c)

    print a.head(1)

def test_stream_of_generator():
    a = pybedtools.example_bedtool('a.bed')
    b = pybedtools.example_bedtool('b.bed')
    b1 = a.intersect(a, stream=True)
    b2 = pybedtools.BedTool((i for i in a)).intersect(a, stream=True)
    sb1 = str(b1)
    sb2 = str(b2)
    print sb1
    print sb2
    assert sb1 == sb2