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
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
|
"""module contains python code for wgcna"""
from unittest import TestCase
from unittest import mock
import pytest
from gn3.computations.wgcna import dump_wgcna_data
from gn3.computations.wgcna import compose_wgcna_cmd
from gn3.computations.wgcna import call_wgcna_script
class TestWgcna(TestCase):
"""test class for wgcna"""
@pytest.mark.unit_test
@mock.patch("gn3.computations.wgcna.process_image")
@mock.patch("gn3.computations.wgcna.run_cmd")
@mock.patch("gn3.computations.wgcna.compose_wgcna_cmd")
@mock.patch("gn3.computations.wgcna.dump_wgcna_data")
def test_call_wgcna_script(self,
mock_dumping_data,
mock_compose_wgcna,
mock_run_cmd,
mock_img,
):
"""test for calling wgcna script"""
# pylint: disable = line-too-long
mock_dumping_data.return_value = "/tmp/QmQPeNsJPyVWPFDVHb77w8G42Fvo15z4bG2X8D2GhfbSXc-test.json"
mock_compose_wgcna.return_value = "Rscript/GUIX_PATH/scripts/r_file.R /tmp/QmQPeNsJPyVWPFDVHb77w8G42Fvo15z4bG2X8D2GhfbSXc-test.json"
request_data = {
"trait_names": ["1455537_at", "1425637_at", "1449593_at", "1421945_a_at", "1450423_s_at", "1423841_at", "1451144_at"],
"trait_sample_data": [
{
"129S1/SvImJ": 7.142,
"A/J": 7.31,
"AKR/J": 7.49,
"B6D2F1": 6.899,
"BALB/cByJ": 7.172,
"BALB/cJ": 7.396
},
{
"129S1/SvImJ": 7.071,
"A/J": 7.05,
"AKR/J": 7.313,
"B6D2F1": 6.999,
"BALB/cByJ": 7.293,
"BALB/cJ": 7.117
}]}
mock_run_cmd_results = {
"code": 0,
"output": "Flagging genes and samples with too many missing values...\n ..step 1\nAllowing parallel execution with up to 3 working processes.\npickSoftThreshold: will use block size 7.\n pickSoftThreshold: calculating connectivity for given powers...\n ..working on genes 1 through 7 of 7\n Flagging genes and samples with too many missing values...\n ..step 1\n ..Working on block 1 .\n TOM calculation: adjacency..\n ..will not use multithreading.\nclustering..\n ....detecting modules..\n ....calculating module eigengenes..\n ....checking kME in modules..\n ..merging modules that are too close..\n mergeCloseModules: Merging modules whose distance is less than 0.15\n mergeCloseModules: less than two proper modules.\n ..color levels are turquoise\n ..there is nothing to merge.\n Calculating new MEs...\n"
}
json_output = "{\"inputdata\":{\"trait_sample_data \":{},\"minModuleSize\":30,\"TOMtype\":\"unsigned\"},\"output\":{\"eigengenes\":[],\"imageLoc\":[],\"colors\":[]}}"
expected_output = {
"data": {
"inputdata": {
"trait_sample_data ": {},
"minModuleSize": 30,
"TOMtype": "unsigned"
},
"output": {
"eigengenes": [],
"imageLoc": [],
"colors": [],
"image_data": "AFDSFNBSDGJJHH"
}
},
**mock_run_cmd_results
}
with mock.patch("builtins.open", mock.mock_open(read_data=json_output)):
mock_run_cmd.return_value = mock_run_cmd_results
mock_img.return_value = b"AFDSFNBSDGJJHH"
results = call_wgcna_script(
"Rscript/GUIX_PATH/scripts/r_file.R", request_data)
mock_dumping_data.assert_called_once_with(request_data)
mock_compose_wgcna.assert_called_once_with(
"Rscript/GUIX_PATH/scripts/r_file.R",
"/tmp/QmQPeNsJPyVWPFDVHb77w8G42Fvo15z4bG2X8D2GhfbSXc-test.json")
mock_run_cmd.assert_called_once_with(
"Rscript/GUIX_PATH/scripts/r_file.R /tmp/QmQPeNsJPyVWPFDVHb77w8G42Fvo15z4bG2X8D2GhfbSXc-test.json")
self.assertEqual(results, expected_output)
@pytest.mark.unit_test
@mock.patch("gn3.computations.wgcna.run_cmd")
@mock.patch("gn3.computations.wgcna.compose_wgcna_cmd")
@mock.patch("gn3.computations.wgcna.dump_wgcna_data")
def test_call_wgcna_script_fails(self, mock_dumping_data, mock_compose_wgcna, mock_run_cmd):
"""test for calling wgcna script\
fails and generates the expected error"""
# pylint: disable = line-too-long,
mock_dumping_data.return_value = "/tmp/QmQPeNsJPyVWPFDVHb77w8G42Fvo15z4bG2X8D2GhfbSXc-test.json"
mock_compose_wgcna.return_value = "Rscript/GUIX_PATH/scripts/r_file.R /tmp/QmQPeNsJPyVWPFDVHb77w8G42Fvo15z4bG2X8D2GhfbSXc-test.json"
expected_error = {
"code": 2,
"output": "could not read the json file"
}
with mock.patch("builtins.open", mock.mock_open(read_data="")):
mock_run_cmd.return_value = expected_error
self.assertEqual(call_wgcna_script(
"input_file.R", ""), expected_error)
@pytest.mark.skip(
"This test assumes that the system will always be invoked from the root"
" of the repository, which is not always true. As such, the code has "
"been updated to always use the absolute paths.")
@pytest.mark.unit_test
def test_compose_wgcna_cmd(self):
"""test for composing wgcna cmd"""
wgcna_cmd = compose_wgcna_cmd(
"wgcna.r", "/tmp/wgcna.json")
self.assertEqual(
wgcna_cmd, "Rscript scripts/wgcna.r /tmp/wgcna.json")
@pytest.mark.unit_test
@mock.patch("gn3.computations.wgcna.TMPDIR", "/tmp")
@mock.patch("gn3.computations.wgcna.uuid.uuid4")
def test_create_json_file(self, file_name_generator):
"""test for writing the data to a csv file"""
# # All the traits we have data for (should not contain duplicates)
# All the strains we have data for (contains duplicates)
trait_sample_data = {"1425642_at": {"129S1/SvImJ": 7.142,
"A/J": 7.31, "AKR/J": 7.49,
"B6D2F1": 6.899, "BALB/cByJ": 7.172,
"BALB/cJ": 7.396},
"1457784_at": {"129S1/SvImJ": 7.071, "A/J": 7.05,
"AKR/J": 7.313,
"B6D2F1": 6.999, "BALB/cByJ": 7.293,
"BALB/cJ": 7.117},
"1444351_at": {"129S1/SvImJ": 7.221, "A/J": 7.246,
"AKR/J": 7.754,
"B6D2F1": 6.866, "BALB/cByJ": 6.752,
"BALB/cJ": 7.269}
}
expected_input = {
"trait_sample_data": trait_sample_data,
"TOMtype": "unsigned",
"minModuleSize": 30
}
with mock.patch("builtins.open", mock.mock_open()) as file_handler:
file_name_generator.return_value = "facb73ff-7eef-4053-b6ea-e91d3a22a00c"
results = dump_wgcna_data(
expected_input)
file_handler.assert_called_once_with(
"/tmp/facb73ff-7eef-4053-b6ea-e91d3a22a00c.json", 'w', encoding='utf-8')
self.assertEqual(
results, "/tmp/facb73ff-7eef-4053-b6ea-e91d3a22a00c.json")
|
