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
|
"""Test cases for procedures defined in commands.py"""
import unittest
from dataclasses import dataclass
from datetime import datetime
from typing import Callable
from unittest import mock
from gn3.commands import compose_gemma_cmd
from gn3.commands import compose_rqtl_cmd
from gn3.commands import queue_cmd
from gn3.commands import run_cmd
from gn3.exceptions import RedisConnectionError
@dataclass
class MockRedis:
"""Mock Redis connection"""
ping: Callable
hset: mock.MagicMock
rpush: mock.MagicMock
class TestCommands(unittest.TestCase):
"""Test cases for commands.py"""
def test_compose_gemma_cmd_no_extra_args(self):
"""Test that the gemma cmd is composed correctly"""
self.assertEqual(
compose_gemma_cmd(gemma_wrapper_cmd="gemma-wrapper",
gemma_kwargs={
"g": "/tmp/genofile.txt",
"p": "/tmp/gf13Ad0tRX/phenofile.txt"
},
gemma_args=["-gk"]),
("gemma-wrapper --json -- "
"-g /tmp/genofile.txt "
"-p /tmp/gf13Ad0tRX/phenofile.txt"
" -gk"))
def test_compose_gemma_cmd_extra_args(self):
"""Test that the gemma cmd is composed correctly"""
self.assertEqual(
compose_gemma_cmd(gemma_wrapper_cmd="gemma-wrapper",
gemma_wrapper_kwargs={
"loco": "1,2,3,4"
},
gemma_kwargs={
"g": "/tmp/genofile.txt",
"p": "/tmp/gf13Ad0tRX/phenofile.txt"
},
gemma_args=["-gk"]),
("gemma-wrapper --json --loco 1,2,3,4 -- "
"-g /tmp/genofile.txt "
"-p /tmp/gf13Ad0tRX/phenofile.txt"
" -gk"))
def test_compose_rqtl_cmd(self):
"""Test that the R/qtl cmd is composed correctly"""
self.assertEqual(
compose_rqtl_cmd(rqtl_wrapper_cmd="rqtl-wrapper",
rqtl_wrapper_kwargs={
"g": "the_genofile",
"p": "the_phenofile",
"model": "np",
"method": "ehk",
"nperm": 2000,
"scale": "Mb",
"control": "rs234567"
},
rqtl_wrapper_bool_kwargs=[
"addcovar"
]),
("Rscript rqtl-wrapper "
"--g the_genofile --p the_phenofile "
"--model np --method ehk "
"--nperm 2000 --scale Mb "
"--control rs234567 "
"--addcovar")
)
def test_queue_cmd_exception_raised_when_redis_is_down(self):
"""Test that the correct error is raised when Redis is unavailable"""
self.assertRaises(RedisConnectionError,
queue_cmd,
cmd="ls",
job_queue="GN2::job-queue",
conn=MockRedis(ping=lambda: False,
hset=mock.MagicMock(),
rpush=mock.MagicMock()))
@mock.patch("gn3.commands.datetime")
@mock.patch("gn3.commands.uuid4")
def test_queue_cmd_correct_calls_to_redis(self, mock_uuid4,
mock_datetime):
"""Test that the cmd is queued properly"""
mock_uuid4.return_value = 1234
mock_datetime.now.return_value = datetime.fromisoformat('2021-02-12 '
'17:32:24.'
'859097')
mock_redis_conn = MockRedis(ping=lambda: True,
hset=mock.MagicMock(),
rpush=mock.MagicMock())
actual_unique_id = "cmd::2021-02-1217-3224-3224-1234"
self.assertEqual(queue_cmd(cmd="ls",
conn=mock_redis_conn,
job_queue="GN2::job-queue"),
actual_unique_id)
mock_redis_conn.hset.assert_has_calls(
[mock.call(name=actual_unique_id, key="cmd", value="ls"),
mock.call(name=actual_unique_id, key="result", value=""),
mock.call(name=actual_unique_id, key="status", value="queued")])
mock_redis_conn.rpush.assert_has_calls(
[mock.call("GN2::job-queue", actual_unique_id)])
@mock.patch("gn3.commands.datetime")
@mock.patch("gn3.commands.uuid4")
def test_queue_cmd_right_calls_to_redis_with_email(self,
mock_uuid4,
mock_datetime):
"""Test that the cmd is queued properly when given the email"""
mock_uuid4.return_value = 1234
mock_datetime.now.return_value = datetime.fromisoformat('2021-02-12 '
'17:32:24.'
'859097')
mock_redis_conn = MockRedis(ping=lambda: True,
hset=mock.MagicMock(),
rpush=mock.MagicMock())
actual_unique_id = "cmd::2021-02-1217-3224-3224-1234"
self.assertEqual(queue_cmd(cmd="ls",
conn=mock_redis_conn,
job_queue="GN2::job-queue",
email="me@me.com"),
actual_unique_id)
mock_redis_conn.hset.assert_has_calls(
[mock.call(name=actual_unique_id, key="cmd", value="ls"),
mock.call(name=actual_unique_id, key="result", value=""),
mock.call(name=actual_unique_id, key="status", value="queued"),
mock.call(name=actual_unique_id, key="email", value="me@me.com")
])
mock_redis_conn.rpush.assert_has_calls(
[mock.call("GN2::job-queue", actual_unique_id)])
def test_run_cmd_correct_input(self):
"""Test that a correct cmd is processed correctly"""
self.assertEqual(run_cmd("echo test"),
{"code": 0, "output": "test\n"})
def test_run_cmd_incorrect_input(self):
"""Test that an incorrect cmd is processed correctly"""
result = run_cmd("echoo test")
self.assertEqual(127, result.get("code"))
self.assertIn("not found", result.get("output"))
|
