1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
|
import unittest
from goby import Reads_pb2, MessageChunks, Alignments_pb2
class CheckChunks(unittest.TestCase):
def testIterate(self):
for chunk in MessageChunks.MessageChunksGenerator(
"../../test-data/compact-reads/s_1_sequence_short.compact-reads"):
collection = Reads_pb2.ReadCollection()
collection.ParseFromString(chunk)
print("chunk has {} reads", len(collection.reads),flush=True)
self.assertTrue(len(collection.reads) > 0)
def testIterateWithType(self):
collection = Reads_pb2.ReadCollection()
for readCollection in MessageChunks.MessageChunksGenerator(
"../../test-data/compact-reads/s_1_sequence_short.compact-reads", collectionContainer=collection):
print("chunk has {} reads", len(readCollection.reads),flush=True)
self.assertTrue(len(readCollection.reads) > 0)
def testIterateAlignment(self):
for chunk in MessageChunks.MessageChunksGenerator("../../test-data/bam/Example.entries"):
collection = Alignments_pb2.AlignmentCollection()
collection.ParseFromString(chunk)
print("chunk has {} entries", len(collection.alignment_entries),flush=True)
self.assertTrue(len(collection.alignment_entries) > 0)
def testIterateAlignmentWithCollection(self):
collection = Alignments_pb2.AlignmentCollection()
for chunk in MessageChunks.MessageChunksGenerator("../../test-data/bam/Example.entries",collection):
print("chunk has {} entries", len(collection.alignment_entries),flush=True)
self.assertTrue(len(collection.alignment_entries) > 0)
if __name__ == '__main__':
unittest.main()
|
