-
Notifications
You must be signed in to change notification settings - Fork 1
/
test_saveAndLoad.py
112 lines (79 loc) · 3.04 KB
/
test_saveAndLoad.py
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
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
import os
import shutil
import unittest
import random
import kProcessor as kp
from params import test_params
class TestSaveAndLoad(unittest.TestCase):
ksize = 21
params_PHMAP = test_params(ksize, "PHMAP")
params_MAP = test_params(ksize, "MAP")
def test_MQF(self):
my_tmpdir = "tmp" + str(random.randint(1, 9999))
os.mkdir(my_tmpdir)
params = test_params(self.ksize, "MQF")
params.Hasher = kp.IntegerHasher(self.ksize)
KF = params.new_kf(self.ksize)
self.assertTrue(KF.empty())
# Create random kmers list
kmers_list = params.generate_kmers(kmers_no=20)
kmers_hashes = [params.Hasher.hash(kmer[0]) for kmer in kmers_list]
for kmer in kmers_list:
KF.insert(kmer[0], kmer[1])
fileName = os.path.join(my_tmpdir, "tmp.kdataframeMQF")
KF.save(fileName)
KF_LOADED = kp.kDataFrame.load(fileName)
it = KF_LOADED.begin()
count = 0
while it != KF_LOADED.end():
self.assertTrue(it.getHashedKmer() in kmers_hashes)
count += 1
it.next()
self.assertEqual(count, len(kmers_list))
shutil.rmtree(my_tmpdir)
def test_MAP(self):
my_tmpdir = "tmp" + str(random.randint(1, 9999))
os.mkdir(my_tmpdir)
params = test_params(self.ksize, "MAP")
params.Hasher = kp.TwoBitsHasher(self.ksize)
KF = params.new_kf(self.ksize)
self.assertTrue(KF.empty())
# Create random kmers list
kmers_list = params.generate_kmers(kmers_no=20)
kmers_hashes = [params.Hasher.hash(kmer[0]) for kmer in kmers_list]
for kmer in kmers_list:
KF.insert(kmer[0], kmer[1])
fileName = os.path.join(my_tmpdir, "tmp.kdataframeMQF")
KF.save(fileName)
KF_LOADED = kp.kDataFrame.load(fileName)
it = KF_LOADED.begin()
count = 0
while it != KF_LOADED.end():
self.assertTrue(it.getHashedKmer() in kmers_hashes)
count += 1
it.next()
self.assertEqual(count, len(kmers_list))
shutil.rmtree(my_tmpdir)
def test_PHMAP(self):
my_tmpdir = "tmp" + str(random.randint(1, 9999))
os.mkdir(my_tmpdir)
params = test_params(self.ksize, "PHMAP")
params.Hasher = kp.TwoBitsHasher(self.ksize)
KF = params.new_kf(self.ksize)
self.assertTrue(KF.empty())
# Create random kmers list
kmers_list = params.generate_kmers(kmers_no=20)
kmers_hashes = [params.Hasher.hash(kmer[0]) for kmer in kmers_list]
for kmer in kmers_list:
KF.insert(kmer[0], kmer[1])
fileName = os.path.join(my_tmpdir, "tmp.kdataframeMQF")
KF.save(fileName)
KF_LOADED = kp.kDataFrame.load(fileName)
it = KF_LOADED.begin()
count = 0
while it != KF_LOADED.end():
self.assertTrue(it.getHashedKmer() in kmers_hashes)
count += 1
it.next()
self.assertEqual(count, len(kmers_list))
shutil.rmtree(my_tmpdir)