|
0
|
1 #!/usr/bin/env python3
|
|
|
2 import unittest
|
|
|
3 from subprocess import PIPE, run
|
|
|
4 import os
|
|
|
5 import shutil
|
|
|
6 import sys
|
|
|
7 import argparse
|
|
|
8
|
|
|
9
|
|
|
10 # This is not best practice but for testing, this is the best I could
|
|
|
11 # come up with
|
|
|
12 sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
|
|
|
13
|
|
|
14 # TODO: Species specific aqquired genes only pheno results, not spec specific?
|
|
|
15
|
|
|
16 test_names = ["test1", "test2", "test3", "test4"]
|
|
|
17 test_data = {
|
|
|
18 # Test published acquired resistance
|
|
|
19 test_names[0]: "data/test_isolate_01.fa",
|
|
|
20 test_names[1]: "data/test_isolate_01_1.fq data/test_isolate_01_2.fq",
|
|
|
21 # Test published point mut resistance
|
|
|
22 test_names[2]: "data/test_isolate_05.fa",
|
|
|
23 test_names[3]: "data/test_isolate_05_1.fq data/test_isolate_05_2.fq",
|
|
|
24 }
|
|
|
25 run_test_dir = "running_test"
|
|
|
26 working_dir = os.path.dirname(os.path.realpath(__file__))
|
|
|
27
|
|
|
28
|
|
|
29 class ResFinderRunTest(unittest.TestCase):
|
|
|
30
|
|
|
31 @classmethod
|
|
|
32 def setUpClass(cls):
|
|
|
33 # Delete "running_test" folder from previous tests if still exists
|
|
|
34 if os.path.isdir(run_test_dir):
|
|
|
35 try:
|
|
|
36 shutil.rmtree(run_test_dir)
|
|
|
37 # The following error has occured using VirtualBox under Windows 10
|
|
|
38 # with ResFinder installed in a shared folder:
|
|
|
39 # OSError [Errno: 26] Text file busy: 'tmp'
|
|
|
40 except OSError:
|
|
|
41 procs = run(["rm", "-r", run_test_dir])
|
|
|
42
|
|
|
43 # Set absolute path for database folders and external programs
|
|
|
44 cls.db_path_res = os.path.abspath(args.db_path_res)
|
|
|
45 cls.blastPath = os.path.abspath(args.blast_path)
|
|
|
46 cls.kmaPath = os.path.abspath(args.kma_path)
|
|
|
47 cls.db_path_point = os.path.abspath(args.db_path_point)
|
|
|
48 cls.dir_res = os.path.join(os.path.dirname(__file__), '../', )
|
|
|
49 cls.dir_res = os.path.abspath(cls.dir_res)
|
|
|
50 # Change working dir to test dir
|
|
|
51 os.chdir(working_dir)
|
|
|
52 # Does not allow running two tests in parallel
|
|
|
53 os.makedirs(run_test_dir, exist_ok=False)
|
|
|
54
|
|
|
55 @classmethod
|
|
|
56 def tearDownClass(cls):
|
|
|
57 try:
|
|
|
58 shutil.rmtree(run_test_dir)
|
|
|
59 # The following error has occured using VirtualBox under Windows 10
|
|
|
60 # with ResFinder installed in a shared folder:
|
|
|
61 # OSError [Errno: 26] Text file busy: 'tmp'
|
|
|
62 except OSError:
|
|
|
63 procs = run(["rm", "-r", run_test_dir])
|
|
|
64
|
|
|
65 def test_on_data_with_just_acquired_resgene_using_blast(self):
|
|
|
66 # Maria has an E. coli isolate, with unknown resistance.
|
|
|
67 # At first, she just wants to know which acquired resistance genes are
|
|
|
68 # found in the genome.
|
|
|
69 # She therefore runs resfinder cmd line.
|
|
|
70
|
|
|
71 # First Maria checks out the documentation.
|
|
|
72 procs = run("python3 ../run_resfinder.py -h", shell=True, stdout=PIPE,
|
|
|
73 check=True)
|
|
|
74 output = procs.stdout.decode()
|
|
|
75 self.assertIn("--help", output)
|
|
|
76
|
|
|
77 # Maria goes on to run ResFinder for acquired genes with her E. coli
|
|
|
78 # isolate.
|
|
|
79 # First she creates a few directories to store her output.
|
|
|
80 test1_dir = run_test_dir + "/" + test_names[0]
|
|
|
81 os.makedirs(test1_dir)
|
|
|
82 # Then she runs run_resfinder with her first isolate.
|
|
|
83 cmd_acquired = ("python3 " + self.dir_res + "/run_resfinder.py"
|
|
|
84 + " -ifa " + test_data[test_names[0]]
|
|
|
85 + " -o " + test1_dir
|
|
|
86 + " -s 'Escherichia coli'"
|
|
|
87 + " --min_cov 0.6"
|
|
|
88 + " -t 0.8"
|
|
|
89 + " --acquired"
|
|
|
90 + " --db_path_res " + self.db_path_res
|
|
|
91 + " --blastPath " + self.blastPath)
|
|
|
92
|
|
|
93 procs = run(cmd_acquired, shell=True, stdout=PIPE, stderr=PIPE,
|
|
|
94 check=True)
|
|
|
95
|
|
|
96 fsa_hit = test1_dir + "/ResFinder_Hit_in_genome_seq.fsa"
|
|
|
97 fsa_res = test1_dir + "/ResFinder_Resistance_gene_seq.fsa"
|
|
|
98 res_table = test1_dir + "/ResFinder_results_table.txt"
|
|
|
99 res_tab = test1_dir + "/ResFinder_results_tab.txt"
|
|
|
100 results = test1_dir + "/ResFinder_results.txt"
|
|
|
101
|
|
|
102 with open(fsa_hit, "r") as fh:
|
|
|
103 check_result = fh.readline()
|
|
|
104 self.assertIn("blaB-2_1_AF189300", check_result)
|
|
|
105
|
|
|
106 with open(fsa_res, "r") as fh:
|
|
|
107 check_result = fh.readline()
|
|
|
108 self.assertIn("blaB-2_AF189300", check_result)
|
|
|
109
|
|
|
110 with open(res_table, "r") as fh:
|
|
|
111 for line in fh:
|
|
|
112 if(line.startswith("blaB-2")):
|
|
|
113 check_result = line
|
|
|
114 break
|
|
|
115 self.assertIn("blaB-2_1_AF189300", check_result)
|
|
|
116
|
|
|
117 with open(res_tab, "r") as fh:
|
|
|
118 fh.readline()
|
|
|
119 check_result = fh.readline()
|
|
|
120 self.assertIn("blaB-2_1_AF189300", check_result)
|
|
|
121
|
|
|
122 with open(results, "r") as fh:
|
|
|
123 fh.readline()
|
|
|
124 fh.readline()
|
|
|
125 fh.readline()
|
|
|
126 fh.readline()
|
|
|
127 fh.readline()
|
|
|
128 check_result = fh.readline()
|
|
|
129 self.assertIn("blaB-2_1_AF189300", check_result)
|
|
|
130
|
|
|
131 def test_on_data_with_just_acquired_resgene_using_kma(self):
|
|
|
132 # Maria has another E. coli isolate, with unknown resistance.
|
|
|
133 # This time she does not have an assembly, but only raw data.
|
|
|
134 # She therefore runs resfinder cmd line using KMA.
|
|
|
135
|
|
|
136 # First she creates a few directories to store her output.
|
|
|
137 test2_dir = run_test_dir + "/" + test_names[1]
|
|
|
138 os.makedirs(test2_dir, exist_ok=False)
|
|
|
139
|
|
|
140 # Then she runs run_resfinder with her first isolate.
|
|
|
141 cmd_acquired = ("python3 " + self.dir_res + "/run_resfinder.py"
|
|
|
142 + " -ifq " + test_data[test_names[1]]
|
|
|
143 + " -o " + test2_dir
|
|
|
144 + " -s 'Escherichia coli'"
|
|
|
145 + " --min_cov 0.6"
|
|
|
146 + " -t 0.8"
|
|
|
147 + " --acquired"
|
|
|
148 + " --db_path_res " + self.db_path_res
|
|
|
149 + " --kmaPath " + self.kmaPath)
|
|
|
150
|
|
|
151 procs = run(cmd_acquired, shell=True, stdout=PIPE, stderr=PIPE,
|
|
|
152 check=True)
|
|
|
153
|
|
|
154 fsa_hit = test2_dir + "/ResFinder_Hit_in_genome_seq.fsa"
|
|
|
155 fsa_res = test2_dir + "/ResFinder_Resistance_gene_seq.fsa"
|
|
|
156 res_table = test2_dir + "/ResFinder_results_table.txt"
|
|
|
157 res_tab = test2_dir + "/ResFinder_results_tab.txt"
|
|
|
158 results = test2_dir + "/ResFinder_results.txt"
|
|
|
159
|
|
|
160 with open(fsa_hit, "r") as fh:
|
|
|
161 check_result = fh.readline()
|
|
|
162 self.assertIn("blaB-2", check_result)
|
|
|
163
|
|
|
164 with open(fsa_res, "r") as fh:
|
|
|
165 check_result = fh.readline()
|
|
|
166 self.assertIn("blaB-2_AF189300", check_result)
|
|
|
167
|
|
|
168 with open(res_table, "r") as fh:
|
|
|
169 for line in fh:
|
|
|
170 if(line.startswith("blaB-2")):
|
|
|
171 check_result = line
|
|
|
172 break
|
|
|
173 self.assertIn("blaB-2", check_result)
|
|
|
174
|
|
|
175 with open(res_tab, "r") as fh:
|
|
|
176 fh.readline()
|
|
|
177 check_result = fh.readline()
|
|
|
178 self.assertIn("blaB-2", check_result)
|
|
|
179
|
|
|
180 with open(results, "r") as fh:
|
|
|
181 fh.readline()
|
|
|
182 fh.readline()
|
|
|
183 fh.readline()
|
|
|
184 fh.readline()
|
|
|
185 fh.readline()
|
|
|
186 check_result = fh.readline()
|
|
|
187 self.assertIn("blaB-2", check_result)
|
|
|
188
|
|
|
189 def test_on_data_with_just_point_mut_using_blast(self):
|
|
|
190 # Maria also wants to check her assembled E. coli isolate for
|
|
|
191 # resistance caused by point mutations.
|
|
|
192
|
|
|
193 # First she creates a few directories to store her output.
|
|
|
194 test3_dir = run_test_dir + "/" + test_names[2]
|
|
|
195 os.makedirs(test3_dir)
|
|
|
196
|
|
|
197 # Then she runs run_resfinder with her first isolate.
|
|
|
198 cmd_point = ("python3 " + self.dir_res + "/run_resfinder.py"
|
|
|
199 + " -ifa " + test_data[test_names[2]]
|
|
|
200 + " -o " + test3_dir
|
|
|
201 + " -s 'Escherichia coli'"
|
|
|
202 + " --min_cov 0.6"
|
|
|
203 + " --threshold 0.8"
|
|
|
204 + " --point"
|
|
|
205 + " --db_path_point " + self.db_path_point
|
|
|
206 + " --db_path_res " + self.db_path_res
|
|
|
207 + " --blastPath " + self.blastPath)
|
|
|
208
|
|
|
209 procs = run(cmd_point, shell=True, stdout=PIPE, stderr=PIPE,
|
|
|
210 check=True)
|
|
|
211
|
|
|
212 # Expected output files
|
|
|
213 pf_pred = test3_dir + "/PointFinder_prediction.txt"
|
|
|
214 pf_res = test3_dir + "/PointFinder_results.txt"
|
|
|
215 pf_table = test3_dir + "/PointFinder_table.txt"
|
|
|
216
|
|
|
217 with open(pf_res, "r") as fh:
|
|
|
218 fh.readline()
|
|
|
219 check_result = fh.readline()
|
|
|
220 self.assertIn("gyrA", check_result)
|
|
|
221 self.assertIn("p.S83A", check_result)
|
|
|
222
|
|
|
223 point_mut_found = False
|
|
|
224 with open(pf_table, "r") as fh:
|
|
|
225 for line in fh:
|
|
|
226 if(line.startswith("gyrA p.S83A")):
|
|
|
227 check_result = line
|
|
|
228 point_mut_found = True
|
|
|
229 break
|
|
|
230 self.assertEqual(point_mut_found, True)
|
|
|
231
|
|
|
232 def test_on_data_with_just_point_mut_using_kma(self):
|
|
|
233 # Maria has another E. coli isolate, with unknown resistance.
|
|
|
234 # This time she does not have an assembly, but only raw data.
|
|
|
235 # She therefore runs resfinder cmd line using KMA.
|
|
|
236
|
|
|
237 # First she creates a few directories to store her output.
|
|
|
238 test4_dir = run_test_dir + "/" + test_names[3]
|
|
|
239 os.makedirs(test4_dir, exist_ok=False)
|
|
|
240
|
|
|
241 # Then she runs run_resfinder with her first isolate.
|
|
|
242 cmd_acquired = ("python3 " + self.dir_res + "/run_resfinder.py"
|
|
|
243 + " -ifq " + test_data[test_names[3]]
|
|
|
244 + " -o " + test4_dir
|
|
|
245 + " -s 'Escherichia coli'"
|
|
|
246 + " --min_cov 0.6"
|
|
|
247 + " --threshold 0.8"
|
|
|
248 + " --point"
|
|
|
249 + " --db_path_point " + self.db_path_point
|
|
|
250 + " --db_path_res " + self.db_path_res
|
|
|
251 + " --kmaPath " + self.kmaPath)
|
|
|
252
|
|
|
253 procs = run(cmd_acquired, shell=True, stdout=PIPE, stderr=PIPE,
|
|
|
254 check=True)
|
|
|
255
|
|
|
256 # Expected output files
|
|
|
257 pf_pred = test4_dir + "/PointFinder_prediction.txt"
|
|
|
258 pf_res = test4_dir + "/PointFinder_results.txt"
|
|
|
259 pf_table = test4_dir + "/PointFinder_table.txt"
|
|
|
260
|
|
|
261 with open(pf_res, "r") as fh:
|
|
|
262 fh.readline()
|
|
|
263 check_result = fh.readline()
|
|
|
264 self.assertIn("gyrA", check_result)
|
|
|
265 self.assertIn("p.S83A", check_result)
|
|
|
266
|
|
|
267 point_mut_found = False
|
|
|
268 with open(pf_table, "r") as fh:
|
|
|
269 for line in fh:
|
|
|
270 if(line.startswith("gyrA p.S83A")):
|
|
|
271 check_result = line
|
|
|
272 point_mut_found = True
|
|
|
273 break
|
|
|
274 self.assertEqual(point_mut_found, True)
|
|
|
275
|
|
|
276
|
|
|
277 def parse_args():
|
|
|
278 parser = argparse.ArgumentParser(add_help=False, allow_abbrev=False)
|
|
|
279 group = parser.add_argument_group("Options")
|
|
|
280 group.add_argument('-res_help', "--resfinder_help",
|
|
|
281 action="help")
|
|
|
282 group.add_argument("-db_res", "--db_path_res",
|
|
|
283 help="Path to the databases for ResFinder",
|
|
|
284 default="./db_resfinder")
|
|
|
285 group.add_argument("-b", "--blastPath",
|
|
|
286 dest="blast_path",
|
|
|
287 help="Path to blastn",
|
|
|
288 default="./cge/blastn")
|
|
|
289 group.add_argument("-k", "--kmaPath",
|
|
|
290 dest="kma_path",
|
|
|
291 help="Path to KMA",
|
|
|
292 default="./cge/kma/kma")
|
|
|
293 group.add_argument("-db_point", "--db_path_point",
|
|
|
294 help="Path to the databases for PointFinder",
|
|
|
295 default="./db_pointfinder")
|
|
|
296 ns, args = parser.parse_known_args(namespace=unittest)
|
|
|
297 return ns, sys.argv[:1] + args
|
|
|
298
|
|
|
299 if __name__ == "__main__":
|
|
|
300 args, argv = parse_args() # run this first
|
|
|
301 sys.argv[:] = argv # create cleans argv for main()
|
|
|
302 unittest.main()
|
