test/test_srf2fastq.py

   1 import os
   2 from StringIO import StringIO
   3 import sys
   4 from unittest import TestCase
   5
   6 _module_path, _module_name = os.path.split(__file__)
   7 sys.path.append(os.path.join(_module_path, '..', 'scripts'))
   8
   9 from htsworkflow.pipelines.test.simulate_runfolder import TESTDATA_DIR
  10
  11 from htsworkflow.pipelines import srf2fastq
  12
  13 class testSrf2Fastq(TestCase):
  14     def test_split_good(self):
  15         source = StringIO("""@header
  16 AGCTTTTT
  17 +
  18 IIIIB+++
  19 """)
  20         target1 = StringIO()
  21         target2 = StringIO()
  22
  23         srf2fastq.convert_single_to_two_fastq(source, target1, target2)
  24
  25         target1.seek(0)
  26         lines1 = target1.readlines()
  27         self.failUnlessEqual(len(lines1),4)
  28         self.failUnlessEqual(lines1[0].rstrip(), '@header/1')
  29         self.failUnlessEqual(lines1[1].rstrip(), 'AGCT')
  30         self.failUnlessEqual(lines1[2].rstrip(), '+')
  31         self.failUnlessEqual(lines1[3].rstrip(), 'IIII')
  32
  33         target2.seek(0)
  34         lines2 = target2.readlines()
  35         self.failUnlessEqual(len(lines2),4)
  36         self.failUnlessEqual(lines2[0].rstrip(), '@header/2')
  37         self.failUnlessEqual(lines2[1].rstrip(), 'TTTT')
  38         self.failUnlessEqual(lines2[2].rstrip(), '+')
  39         self.failUnlessEqual(lines2[3].rstrip(), 'B+++')
  40
  41     def test_split_at_with_header(self):
  42         source = StringIO("""@header1
  43 AGCTTTTT
  44 +
  45 @IIIB+++
  46 @header2
  47 AGCTTTTT
  48 +
  49 IIIIB+++
  50 """)
  51         target1 = StringIO()
  52         target2 = StringIO()
  53
  54         srf2fastq.convert_single_to_two_fastq(source, target1, target2, header="foo_")
  55
  56         target1.seek(0)
  57         lines1 = target1.readlines()
  58         self.failUnlessEqual(len(lines1),8)
  59         self.failUnlessEqual(lines1[0].rstrip(), '@foo_header1/1')
  60         self.failUnlessEqual(lines1[1].rstrip(), 'AGCT')
  61         self.failUnlessEqual(lines1[2].rstrip(), '+')
  62         self.failUnlessEqual(lines1[3].rstrip(), '@III')
  63
  64         target2.seek(0)
  65         lines2 = target2.readlines()
  66         self.failUnlessEqual(len(lines2),8)
  67         self.failUnlessEqual(lines2[0].rstrip(), '@foo_header1/2')
  68         self.failUnlessEqual(lines2[1].rstrip(), 'TTTT')
  69         self.failUnlessEqual(lines2[2].rstrip(), '+')
  70         self.failUnlessEqual(lines2[3].rstrip(), 'B+++')
  71
  72     def test_single_at(self):
  73         source = StringIO("""@header1
  74 AGCTTTTT
  75 +
  76 @IIIB+++
  77 @header2
  78 AGCTTTTT
  79 +
  80 IIIIB+++
  81 """)
  82         target1 = StringIO()
  83
  84         srf2fastq.convert_single_to_fastq(source, target1)
  85
  86         target1.seek(0)
  87         lines1 = target1.readlines()
  88         self.failUnlessEqual(len(lines1),8)
  89         self.failUnlessEqual(lines1[0].rstrip(), '@header1')
  90         self.failUnlessEqual(lines1[1].rstrip(), 'AGCTTTTT')
  91         self.failUnlessEqual(lines1[2].rstrip(), '+')
  92         self.failUnlessEqual(lines1[3].rstrip(), '@IIIB+++')
  93
  94     def test_single_at_with_header(self):
  95         source = StringIO("""@header1
  96 AGCTTTTT
  97 +
  98 @IIIB+++
  99 @header2
 100 AGCTTTTT
 101 +
 102 IIIIB+++
 103 """)
 104         target1 = StringIO()
 105
 106         srf2fastq.convert_single_to_fastq(source, target1, header="foo_")
 107
 108         target1.seek(0)
 109         lines1 = target1.readlines()
 110         self.failUnlessEqual(len(lines1),8)
 111         self.failUnlessEqual(lines1[0].rstrip(), '@foo_header1')
 112         self.failUnlessEqual(lines1[1].rstrip(), 'AGCTTTTT')
 113         self.failUnlessEqual(lines1[2].rstrip(), '+')
 114         self.failUnlessEqual(lines1[3].rstrip(), '@IIIB+++')
 115
 116     def test_is_srf(self):
 117         cnf4_srf = 'woldlab_070829_USI-EAS44_0017_FC11055_1.srf'
 118         cnf4_path = os.path.join(TESTDATA_DIR, cnf4_srf)
 119         cnf1_srf = 'woldlab_090512_HWI-EAS229_0114_428NNAAXX_5.srf'
 120         cnf1_path = os.path.join(TESTDATA_DIR, cnf1_srf)
 121
 122         is_srf = srf2fastq.is_srf
 123         self.failUnlessEqual(is_srf(__file__), False)
 124         self.failUnlessEqual(is_srf(cnf4_path), True)
 125         self.failUnlessEqual(is_srf(cnf1_path), True)
 126
 127     def test_is_cnf1(self):
 128         cnf4_srf = 'woldlab_070829_USI-EAS44_0017_FC11055_1.srf'
 129         cnf4_path = os.path.join(TESTDATA_DIR, cnf4_srf)
 130         cnf1_srf = 'woldlab_090512_HWI-EAS229_0114_428NNAAXX_5.srf'
 131         cnf1_path = os.path.join(TESTDATA_DIR, cnf1_srf)
 132
 133         is_cnf1 = srf2fastq.is_cnf1
 134         self.failUnlessRaises(ValueError, is_cnf1, __file__)
 135         self.failUnlessEqual(is_cnf1(cnf4_path), False)
 136         self.failUnlessEqual(is_cnf1(cnf1_path), True)
 137
 138
 139 def suite():
 140     from unittest import TestSuite, defaultTestLoader
 141     suite = TestSuite()
 142     suite.addTests(defaultTestLoader.loadTestsFromTestCase(testSrf2Fastq))
 143     return suite
 144
 145
 146 if __name__ == "__main__":
 147     from unittest import main
 148     main(defaultTest="suite")