workflow/tests/test_trim_reads.py

+#!/usr/bin/env python3
+
+import pytest
+import os
+
+test_data_path = os.path.dirname(os.path.abspath(__file__)) + \
+                '/../../test_data/'
+test_output_path = os.path.dirname(os.path.abspath(__file__)) + \
+                '/../output/trimReads/'
+
+
+
+@pytest.mark.singleend
+def test_trim_reads_singleend():
+    raw_fastq = test_data_path + 'ENCFF833BLU.fastq.gz'
+    trimmed_fastq = test_output_path + 'ENCLB144FDT/ENCLB144FDT_R1_trimmed.fq.gz'
+    assert os.path.getsize(raw_fastq) != os.path.getsize(trimmed_fastq)
+    assert os.path.getsize(trimmed_fastq) == 2512853101
+
+
+@pytest.mark.singleend
+def test_trim_report_singleend():
+    trimmed_fastq_report = test_output_path + \
+                            'ENCLB144FDT/ENCLB144FDT_R1.fastq.gz_trimming_report.txt'
+    assert 'Trimming mode: single-end' in open(trimmed_fastq_report).readlines()[4]
+
+
+@pytest.mark.pairedend
+def test_trim_reads_pairedend():
+    raw_fastq = test_data_path + 'ENCFF582IOZ.fastq.gz'
+    trimmed_fastq = test_output_path + 'ENCLB637LZP/ENCLB637LZP_R2_val_2.fq.gz'
+    assert os.path.getsize(raw_fastq) != os.path.getsize(trimmed_fastq)
+    assert os.path.getsize(trimmed_fastq) == 2229312710
+
+
+@pytest.mark.pairedend
+def test_trim_report_pairedend():
+    trimmed_fastq_report = test_output_path + \
+                            'ENCLB637LZP/ENCLB637LZP_R2.fastq.gz_trimming_report.txt'
+    assert 'Trimming mode: paired-end' in open(trimmed_fastq_report).readlines()[4]

workflow/tests/test_utils.py

+#!/usr/bin/env python3
+
+import pytest
+import utils
+
+
+STRIP_EXTENSIONS = ['.gz', '.fq', '.fastq', '.fa', '.fasta']
+
+
+@pytest.fixture
+def steps():
+    steps = []
+    return steps
+
+
+@pytest.fixture
+def steps_1(steps):
+    design_file = "test_data/design_ENCSR238SGC_SE.txt"
+    step = [
+        "grep H3K4me1 %s " % (design_file)]
+    return step
+
+
+@pytest.fixture
+def steps_2(steps_1):
+    steps_1.extend([
+        "cut -f8"
+    ])
+    return steps_1
+
+
+@pytest.mark.unit
+def test_run_one_step(steps_1, capsys):
+    check_output = 'ENCLB144FDT\tENCSR238SGC\tlimb\tH3K4me1\tNone\t1\tENCLB304SBJ\tENCFF833BLU.fastq.gz'.encode('UTF-8')
+    out, err = utils.run_pipe(steps_1)
+    output, errors = capsys.readouterr()
+    assert "first step shlex to stdout" in output
+    assert check_output in out
+
+
+@pytest.mark.unit
+def test_run_two_step(steps_2, capsys):
+    check_output = 'ENCFF833BLU.fastq.gz\nENCFF646LXU.fastq.gz'.encode('UTF-8')
+    out, err = utils.run_pipe(steps_2)
+    output, errors = capsys.readouterr()
+    assert "intermediate step 2 shlex to stdout" in output
+    assert check_output in out
+
+
+@pytest.mark.unit
+def test_run_last_step_file(steps_2, capsys, tmpdir):
+    check_output = 'ENCFF833BLU.fastq.gz\nENCFF646LXU.fastq.gz'
+    tmp_outfile = tmpdir.join('output.txt')
+    out, err = utils.run_pipe(steps_2, tmp_outfile.strpath)
+    output, errors = capsys.readouterr()
+    assert "last step shlex" in output
+    assert check_output in tmp_outfile.read()
+
+
+@pytest.mark.unit
+def test_strip_extensions():
+    filename = utils.strip_extensions('ENCFF833BLU.fastq.gz', STRIP_EXTENSIONS)
+    assert filename == 'ENCFF833BLU'
+
+
+@pytest.mark.unit
+def test_strip_extensions_not_valid():
+    filename = utils.strip_extensions('ENCFF833BLU.not.valid', STRIP_EXTENSIONS)
+    assert filename == 'ENCFF833BLU.not.valid'
+
+
+@pytest.mark.unit
+def test_strip_extensions_missing_basename():
+    filename = utils.strip_extensions('.fastq.gz', STRIP_EXTENSIONS)
+    assert filename == '.fastq'

workflow/tests/test_xcor.py

+#!/usr/bin/env python3
+
+import pytest
+import os
+import pandas as pd
+
+test_output_path = os.path.dirname(os.path.abspath(__file__)) + \
+                '/../output/crossReads/'
+
+
+@pytest.mark.singleend
+def test_cross_plot_singleend():
+    assert os.path.exists(os.path.join(test_output_path, 'ENCLB144FDT/ENCLB144FDT.cc.plot.pdf'))
+
+
+@pytest.mark.singleend
+def test_cross_qc_singleend():
+    qc_file = os.path.join(test_output_path,"ENCLB144FDT/ENCLB144FDT.cc.qc")
+    df_xcor = pd.read_csv(qc_file, sep="\t", header=None)
+    assert df_xcor[2].iloc[0] == '185,195,205'
+    assert df_xcor[8].iloc[0] == 1.02454
+    assert df_xcor[9].iloc[0] == 0.8098014
+
+
+@pytest.mark.pairedend
+def test_cross_qc_pairedend():
+    assert os.path.exists(os.path.join(test_output_path, 'ENCLB568IYX/ENCLB568IYX.cc.plot.pdf'))
+
+
+@pytest.mark.pairedend
+def test_cross_plot_pairedend():
+    qc_file = os.path.join(test_output_path,"ENCLB568IYX/ENCLB568IYX.cc.qc")
+    df_xcor = pd.read_csv(qc_file, sep="\t", header=None)
+    assert df_xcor[2].iloc[0] == '215,225,455'
+    assert round(df_xcor[8].iloc[0],6) == 1.056201
+    assert df_xcor[9].iloc[0] == 3.599357