2 from StringIO import StringIO
6 _module_path, _module_name = os.path.split(__file__)
7 sys.path.append(os.path.join(_module_path, '..', 'scripts'))
9 from htsworkflow.pipelines.test.simulate_runfolder import TESTDATA_DIR
11 from htsworkflow.pipelines import srf2fastq
13 class testSrf2Fastq(unittest.TestCase):
14 def test_split_good(self):
15 source = StringIO("""@header
23 srf2fastq.convert_single_to_two_fastq(source, target1, target2)
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')
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+++')
41 def test_split_at_with_header(self):
42 source = StringIO("""@header1
54 srf2fastq.convert_single_to_two_fastq(source, target1, target2, header="foo_")
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')
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+++')
72 def test_single_at(self):
73 source = StringIO("""@header1
84 srf2fastq.convert_single_to_fastq(source, target1)
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+++')
94 def test_single_at_with_header(self):
95 source = StringIO("""@header1
106 srf2fastq.convert_single_to_fastq(source, target1, header="foo_")
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+++')
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)
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)
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)
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)
139 return unittest.makeSuite(testSrf2Fastq,'test')
141 if __name__ == "__main__":
142 unittest.main(defaultTest="suite")