2 """More direct synthetic test cases for the eland output file processing
4 from StringIO import StringIO
7 from htsworkflow.pipelines.eland import ElandLane
9 class ElandTests(unittest.TestCase):
10 """Test specific Eland modules
12 def compare_match_array(self, current, expected):
13 for key in expected.keys():
14 self.failUnlessEqual(current[key], expected[key],
15 "Key %s: %s != %s" % (key,current[key],expected[key]))
17 def test_eland_score_mapped_mismatches(self):
19 match_codes = {'NM':0, 'QC':0, 'RM':0,
20 'U0':0, 'U1':0, 'U2':0,
21 'R0':0, 'R1':0, 'R2':0,
23 r = eland._score_mapped_mismatches("QC", match_codes)
24 self.failUnlessEqual(r, ElandLane.SCORE_QC)
25 self.compare_match_array(match_codes,
26 {'NM':0, 'QC':1, 'RM':0,
27 'U0':0, 'U1':0, 'U2':0,
28 'R0':0, 'R1':0, 'R2':0,
31 r = eland._score_mapped_mismatches("NM", match_codes)
32 self.failUnlessEqual(r, ElandLane.SCORE_QC)
33 self.compare_match_array(match_codes,
34 {'NM':1, 'QC':1, 'RM':0,
35 'U0':0, 'U1':0, 'U2':0,
36 'R0':0, 'R1':0, 'R2':0,
39 r = eland._score_mapped_mismatches("1:0:0", match_codes)
40 self.failUnlessEqual(r, ElandLane.SCORE_READ)
41 self.compare_match_array(match_codes,
42 {'NM':1, 'QC':1, 'RM':0,
43 'U0':1, 'U1':0, 'U2':0,
44 'R0':0, 'R1':0, 'R2':0,
47 r = eland._score_mapped_mismatches("2:4:16", match_codes)
48 self.failUnlessEqual(r, ElandLane.SCORE_READ)
49 self.compare_match_array(match_codes,
50 {'NM':1, 'QC':1, 'RM':0,
51 'U0':1, 'U1':0, 'U2':0,
52 'R0':2, 'R1':4, 'R2':16,
55 r = eland._score_mapped_mismatches("1:1:1", match_codes)
56 self.failUnlessEqual(r, ElandLane.SCORE_READ)
57 self.compare_match_array(match_codes,
58 {'NM':1, 'QC':1, 'RM':0,
59 'U0':2, 'U1':1, 'U2':1,
60 'R0':2, 'R1':4, 'R2':16,
63 r = eland._score_mapped_mismatches("1:0:0", match_codes)
64 self.failUnlessEqual(r, ElandLane.SCORE_READ)
65 self.compare_match_array(match_codes,
66 {'NM':1, 'QC':1, 'RM':0,
67 'U0':3, 'U1':1, 'U2':1,
68 'R0':2, 'R1':4, 'R2':16,
71 r = eland._score_mapped_mismatches("0:0:1", match_codes)
72 self.failUnlessEqual(r, ElandLane.SCORE_READ)
73 self.compare_match_array(match_codes,
74 {'NM':1, 'QC':1, 'RM':0,
75 'U0':3, 'U1':1, 'U2':2,
76 'R0':2, 'R1':4, 'R2':16,
79 r = eland._score_mapped_mismatches("chr3.fa", match_codes)
80 self.failUnlessEqual(r, ElandLane.SCORE_UNRECOGNIZED)
81 self.compare_match_array(match_codes,
82 {'NM':1, 'QC':1, 'RM':0,
83 'U0':3, 'U1':1, 'U2':2,
84 'R0':2, 'R1':4, 'R2':16,
87 def test_count_mapped_export(self):
90 r = eland._count_mapped_export(mapped_reads, "chr3.fa", "38")
91 self.failUnlessEqual(mapped_reads['chr3.fa'], 1)
92 self.failUnlessEqual(r, 'U0')
95 r = eland._count_mapped_export(mapped_reads, "chr3.fa", "36A4")
96 self.failUnlessEqual(mapped_reads['chr3.fa'], 1)
97 self.failUnlessEqual(r, 'U1')
100 r = eland._count_mapped_export(mapped_reads, "chr3.fa", "30A2T2")
101 self.failUnlessEqual(mapped_reads['chr3.fa'], 1)
102 self.failUnlessEqual(r, 'U2')
105 r = eland._count_mapped_export(mapped_reads, "chr3.fa", "26AG2T2")
106 self.failUnlessEqual(mapped_reads['chr3.fa'], 1)
107 self.failUnlessEqual(r, 'U2')
111 r = eland._count_mapped_export(mapped_reads, "chr3.fa", "26^AG$4")
112 self.failUnlessEqual(mapped_reads['chr3.fa'], 1)
113 self.failUnlessEqual(r, 'U2')
117 r = eland._count_mapped_export(mapped_reads, "chr3.fa", "26^2$4")
118 self.failUnlessEqual(mapped_reads['chr3.fa'], 1)
119 self.failUnlessEqual(r, 'U0')
121 def test_update_eland_export(self):
122 """Test scoring the pipeline export file"""
124 qc_read = StringIO("ILLUMINA-33A494 1 1 1 3291 1036 0 1 GANNTCCTCACCCGACANNNNNNNANNNCGGGNNACTC \XBB]^^^^[`````BBBBBBBBBBBBBBBBBBBBBBB QC")
125 one_read_exact = StringIO("ILLUMINA-33A494 1 1 1 2678 1045 0 1 AAGGTGAAGAAGGAGATGNNGATGATGACGACGATAGA ]]WW[[W]W]]R\WWZ[RBBS^\XVa____]W[]]___ chrX.fa 148341829 F 38 45")
126 one_read_mismatch = StringIO("ILLUMINA-33A494 1 1 1 2678 1045 0 1 AAGGTGAAGAAGGAGATGNNGATGATGACGACGATAGA ]]WW[[W]W]]R\WWZ[RBBS^\XVa____]W[]]___ chrX.fa 148341829 F 18AA15G1T 45")
127 multi_read = StringIO("ILLUMINA-33A494 1 1 1 4405 1046 0 1 GTGGTTTCGCTGGATAGTNNGTAGGGACAGTGGGAATC ``````````__a__V^XBB^SW^^a_____a______ 9:2:1")
129 match_codes, match_reads, reads = eland._update_eland_export(qc_read)
130 self.compare_match_array(match_codes,
131 {'NM':0, 'QC':1, 'RM':0,
132 'U0':0, 'U1':0, 'U2':0,
133 'R0':0, 'R1':0, 'R2':0,
135 self.failUnlessEqual(len(match_reads), 0)
136 self.failUnlessEqual(reads, 1)
138 match_codes, match_reads, reads = eland._update_eland_export(one_read_exact)
139 self.compare_match_array(match_codes,
140 {'NM':0, 'QC':0, 'RM':0,
141 'U0':1, 'U1':0, 'U2':0,
142 'R0':0, 'R1':0, 'R2':0,
144 self.failUnlessEqual(match_reads['chrX.fa'], 1)
145 self.failUnlessEqual(reads, 1)
147 match_codes, match_reads, reads = eland._update_eland_export(one_read_mismatch)
148 self.compare_match_array(match_codes,
149 {'NM':0, 'QC':0, 'RM':0,
150 'U0':0, 'U1':0, 'U2':1,
151 'R0':0, 'R1':0, 'R2':0,
153 self.failUnlessEqual(match_reads['chrX.fa'], 1)
154 self.failUnlessEqual(reads, 1)
156 match_codes, match_reads, reads = eland._update_eland_export(multi_read)
157 self.compare_match_array(match_codes,
158 {'NM':0, 'QC':0, 'RM':0,
159 'U0':0, 'U1':0, 'U2':1,
160 'R0':9, 'R1':2, 'R2':0,
162 self.failUnlessEqual(len(match_reads), 0)
163 self.failUnlessEqual(reads, 1)
167 return unittest.makeSquite(ElandTests, 'test')
169 if __name__ == "__main__":
170 unittest.main(defaultTest="suite")