Mercurial > repos > dcouvin > resfinder4

comparison resfinder/tests/functional_tests.py @ 0:55051a9bc58d draft default tip

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Uploaded
author dcouvin
date Mon, 10 Jan 2022 20:06:07 +0000
parents
children
comparison
equal deleted inserted replaced
-1:000000000000 0:55051a9bc58d
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()