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
|
"""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 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_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("cmd", "ls", actual_unique_id),
mock.call("result", "", actual_unique_id),
mock.call("status", "queued", actual_unique_id)])
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("cmd", "ls", actual_unique_id),
mock.call("result", "", actual_unique_id),
mock.call("status", "queued", actual_unique_id),
mock.call("email", "me@me.com", actual_unique_id)])
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"))
|
