|
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 from collections import namedtuple
|
|
|
8
|
|
|
9 sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
|
|
|
10
|
|
|
11 from cgecore.cgefinder import CGEFinder
|
|
|
12 from cge.resfinder import ResFinder
|
|
|
13 from cge.pointfinder import PointFinder
|
|
|
14
|
|
|
15 run_test_dir = "running_test"
|
|
|
16 working_dir = os.path.dirname(os.path.realpath(__file__))
|
|
|
17
|
|
|
18 test_names = ["test1", "test2", "test3", "test4"]
|
|
|
19
|
|
|
20
|
|
|
21 class TestFinder(CGEFinder):
|
|
|
22
|
|
|
23 def __init__(results):
|
|
|
24 """
|
|
|
25 Marias minimalistic Finder class
|
|
|
26 """
|
|
|
27 self.results = results
|
|
|
28
|
|
|
29 def resistance_analysis_genes(self):
|
|
|
30 """
|
|
|
31 Running some analysis that yield results appropriate for resistance
|
|
|
32 type results
|
|
|
33 """
|
|
|
34 Gene = namedtuple("Gene",
|
|
|
35 ["name", "length", "aln_ln", "aln_id", "aln_gaps",
|
|
|
36 "name_variant", "acc", "query", "q_start", "q_end",
|
|
|
37 "q_depth", "t_start", "t_end"])
|
|
|
38
|
|
|
39 analysis_results = {
|
|
|
40 "group1": {
|
|
|
41 "gene1": None, "gene2": None},
|
|
|
42 "group2": {
|
|
|
43 "gene3": None},
|
|
|
44 "group3": {},
|
|
|
45 "group4": {
|
|
|
46 "gene1": None}
|
|
|
47 }
|
|
|
48
|
|
|
49 analysis_results["group1"]["gene1"] = Gene(
|
|
|
50 "gene1", 1000, 890, 0.97, 2, "gene1b", "NC_SOMEGENE", "contig12",
|
|
|
51 24, 1024, 50, 1, 1000)
|
|
|
52 analysis_results["group1"]["gene2"] = Gene(
|
|
|
53 "gene2", 500, 500, 1, 0, None, "NC_SOMEGENE2", "contig3",
|
|
|
54 24, 1024, None, 1, 500)
|
|
|
55 analysis_results["group2"]["gene3"] = Gene(
|
|
|
56 "gene3", 5000, 4500, 0.85, 10, None, "NC_SOMEGENE3", None,
|
|
|
57 None, None, 50, 1, 5000)
|
|
|
58 analysis_results["group4"]["gene1"] = Gene(
|
|
|
59 "gene1", 1000, 890, 0.97, 2, "gene1b", "NC_SOMEGENE", "contig12",
|
|
|
60 24, 1024, 49.74, 1, 1000)
|
|
|
61
|
|
|
62 for group in analysis_results:
|
|
|
63 for name, gene in group.items():
|
|
|
64 self.results.add_gene(
|
|
|
65 template_name=gene.name,
|
|
|
66 template_length=gene.length,
|
|
|
67 template_start_pos=gene.t_start,
|
|
|
68 templare_end_pos=gene.t_end,
|
|
|
69 aln_length=gene.aln_ln,
|
|
|
70 aln_identity=gene.aln_id,
|
|
|
71 aln_gaps=gene.aln_gaps,
|
|
|
72 query_id=gene.query,
|
|
|
73 query_start_pos=gene.q_start,
|
|
|
74 query_end_pos=gene.q_end,
|
|
|
75 query_depth=gene.q_depth
|
|
|
76 )
|
|
|
77
|
|
|
78 def resistance_analysis_mutations(self):
|
|
|
79 """
|
|
|
80 Running some analysis that yield results appropriate for resistance
|
|
|
81 type results
|
|
|
82 """
|
|
|
83 pass
|
|
|
84
|
|
|
85 def resistance_analysis_phenotypes(self):
|
|
|
86 """
|
|
|
87 Running some analysis that yield results appropriate for resistance
|
|
|
88 type results
|
|
|
89 """
|
|
|
90 pass
|
|
|
91
|
|
|
92
|
|
|
93 class ResFinderRunTest(unittest.TestCase):
|
|
|
94
|
|
|
95 def setUp(self):
|
|
|
96 # Change working dir to test dir
|
|
|
97 os.chdir(working_dir)
|
|
|
98 # Does not allow running two tests in parallel
|
|
|
99 os.makedirs(run_test_dir, exist_ok=False)
|
|
|
100
|
|
|
101 def tearDown(self):
|
|
|
102 shutil.rmtree(run_test_dir)
|
|
|
103
|
|
|
104 def test_create_and_store_results_in_resistance_type(self):
|
|
|
105 """
|
|
|
106 Maria has created a CGE service. The service use the cge_core_module
|
|
|
107 to create a CGEFinder object, then runs some analysis which stores its
|
|
|
108 results using the output module.
|
|
|
109 """
|
|
|
110 # Maria records when the script is being run
|
|
|
111 run_start = time.gmtime(time.time())
|
|
|
112 run_start_cge = ",".join(run_start[0], run_start[1], run_start[2],
|
|
|
113 run_start[3], run_start[4], run_start[5])
|
|
|
114
|
|
|
115 # First she creates a few directories to store her output.
|
|
|
116 test1_dir = run_test_dir + "/" + test_names[0]
|
|
|
117 os.makedirs(test1_dir)
|
|
|
118
|
|
|
119 # Create
|
|
|
120 results = Resistance(date=run_start_cge, name="TestFinder", "0.01", "Fake_ID", dbs)
|
|
|
121 finder = TestFinder()
|
|
|
122 finder.resistance_analysis_genes()
|
|
|
123 finder.resistance_analysis_mutations()
|
|
|
124 finder.resistance_analysis_phenotypes()
|
|
|
125 out_str = results.dump_json()
|
|
|
126 results.dump_json("{}/data.json".format(test1_dir))
|
