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
|
"""Test cases for procedures defined in commands.py"""
import os
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"""
@mock.patch("gn3.commands.lookup_file")
def test_compose_gemma_cmd_no_extra_args(self, mock_lookup_file):
"""Test that thhe gemma cmd is composed correctly"""
metadata_file = os.path.join(os.path.dirname(__file__),
"test_data/metadata.json")
mock_lookup_file.side_effect = [metadata_file,
"/tmp/genofile.txt",
"/tmp/gf13Ad0tRX/phenofile.txt"]
self.assertEqual(compose_gemma_cmd("gf13Ad0t",
"metadata.json",
gemma_wrapper_cmd="gemma-wrapper",
gemma_wrapper_kwargs=None,
gemma_kwargs=None,
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,
"ls",
MockRedis(ping=lambda: False,
hset=mock.MagicMock(),
rpush=mock.MagicMock()))
@mock.patch("gn3.commands.datetime")
@mock.patch("gn3.commands.uuid4")
def test_queue_cmd_right_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("ls", mock_redis_conn), 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)])
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"))
|
