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
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
|
"""Test cases for gn3.db.case_attributes.py"""
import pytest
import pickle
import tempfile
import os
import json
from pathlib import Path
from pytest_mock import MockFixture
from gn3.db.case_attributes import queue_edit
from gn3.db.case_attributes import (
CaseAttributeEdit,
EditStatus,
apply_change,
view_change
)
@pytest.mark.unit_test
def test_queue_edit(mocker: MockFixture) -> None:
mock_conn = mocker.MagicMock()
with mock_conn.cursor() as cursor:
type(cursor).lastrowid = 28
tmpdir = os.environ.get("TMPDIR", tempfile.gettempdir())
caseattr_id = queue_edit(
cursor,
directory=tmpdir,
edit=CaseAttributeEdit(
inbredset_id=1, status=EditStatus.review,
user_id="xxxx", changes={"a": 1, "b": 2}
))
cursor.execute.assert_called_once_with(
"INSERT INTO "
"caseattributes_audit(status, editor, json_diff_data) "
"VALUES (%s, %s, %s) "
"ON DUPLICATE KEY UPDATE status=%s",
('review', 'xxxx', '{"a": 1, "b": 2}', 'review'))
assert 28 == caseattr_id
@pytest.mark.unit_test
def test_view_change(mocker: MockFixture) -> None:
"""Test view_change function."""
sample_json_diff = {
"inbredset_id": 1,
"Modifications": {
"Original": {
"B6D2F1": {"Epoch": "10au"},
"BXD100": {"Epoch": "3b"},
"BXD101": {"SeqCvge": "29"},
"BXD102": {"Epoch": "3b"},
"BXD108": {"SeqCvge": ""}
},
"Current": {
"B6D2F1": {"Epoch": "10"},
"BXD100": {"Epoch": "3"},
"BXD101": {"SeqCvge": "2"},
"BXD102": {"Epoch": "3"},
"BXD108": {"SeqCvge": "oo"}
}
}
}
change_id = 28
mock_cursor, mock_conn = mocker.MagicMock(), mocker.MagicMock()
mock_conn.cursor.return_value = mock_cursor
mock_cursor.fetchone.return_value = (json.dumps(sample_json_diff), None)
assert view_change(mock_cursor, change_id) == sample_json_diff
mock_cursor.execute.assert_called_once_with(
"SELECT json_diff_data FROM caseattributes_audit WHERE id = %s",
(change_id,))
mock_cursor.fetchone.assert_called_once()
@pytest.mark.unit_test
def test_view_change_invalid_json(mocker: MockFixture) -> None:
"""Test invalid json when view_change is called"""
change_id = 28
mock_cursor, mock_conn = mocker.MagicMock(), mocker.MagicMock()
mock_conn.cursor.return_value = mock_cursor
mock_cursor.fetchone.return_value = ("invalid_json_string", None)
with pytest.raises(json.JSONDecodeError):
view_change(mock_cursor, change_id)
mock_cursor.execute.assert_called_once_with(
"SELECT json_diff_data FROM caseattributes_audit WHERE id = %s",
(change_id,))
@pytest.mark.unit_test
def test_view_change_no_data(mocker: MockFixture) -> None:
"Test no result when view_change is called"
change_id = 28
mock_cursor, mock_conn = mocker.MagicMock(), mocker.MagicMock()
mock_cursor.fetchone.return_value = (None, None)
assert view_change(mock_cursor, change_id) == {}
mock_cursor.execute.assert_called_once_with(
"SELECT json_diff_data FROM caseattributes_audit WHERE id = %s",
(change_id,))
@pytest.mark.unit_test
def test_apply_change_approved(mocker: MockFixture) -> None:
"""Test approving a change"""
mock_cursor, mock_conn = mocker.MagicMock(), mocker.MagicMock()
mock_conn.cursor.return_value = mock_cursor
mock_lmdb = mocker.patch("gn3.db.case_attributes.lmdb")
mock_env, mock_txn = mocker.MagicMock(), mocker.MagicMock()
mock_lmdb.open.return_value = mock_env
mock_env.begin.return_value.__enter__.return_value = mock_txn
change_id, review_ids = 1, {1, 2, 3}
mock_txn.get.side_effect = (
pickle.dumps(review_ids), # b"review" key
None, # b"approved" key
)
tmpdir = Path(os.environ.get("TMPDIR", tempfile.gettempdir()))
mock_cursor.fetchone.return_value = (json.dumps({
"inbredset_id": 1,
"Modifications": {
"Current": {
"B6D2F1": {"Epoch": "10"},
"BXD100": {"Epoch": "3"},
"BXD101": {"SeqCvge": "2"},
"BXD102": {"Epoch": "3"},
"BXD108": {"SeqCvge": "oo"}
}
}
}), None)
mock_cursor.fetchall.side_effect = [
[ # Strain query
("B6D2F1", 1), ("BXD100", 2),
("BXD101", 3), ("BXD102", 4),
("BXD108", 5)],
[ # CaseAttribute query
("Epoch", 101), ("SeqCvge", 102)]
]
assert apply_change(mock_cursor, EditStatus.approved,
change_id, tmpdir) is True
assert mock_cursor.execute.call_count == 4
mock_cursor.execute.assert_has_calls([
mocker.call(
"SELECT json_diff_data FROM caseattributes_audit WHERE id = %s",
(change_id,)),
mocker.call(
"SELECT Name, Id FROM Strain WHERE Name IN (%s, %s, %s, %s, %s)",
("B6D2F1", "BXD100", "BXD101", "BXD102", "BXD108")),
mocker.call(
"SELECT Name, CaseAttributeId FROM CaseAttribute "
"WHERE InbredSetId = %s AND Name IN (%s, %s)",
(1, "SeqCvge", "Epoch")),
mocker.call(
"UPDATE caseattributes_audit SET status = %s WHERE id = %s",
("approved", change_id))
])
mock_cursor.executemany.assert_called_once_with(
"INSERT INTO CaseAttributeXRefNew (InbredSetId, StrainId, CaseAttributeId, Value) "
"VALUES (%(inbredset_id)s, %(strain_id)s, %(caseattr_id)s, %(value)s) "
"ON DUPLICATE KEY UPDATE Value = VALUES(Value)",
[
{"inbredset_id": 1, "strain_id": 1, "caseattr_id": 101, "value": "10"},
{"inbredset_id": 1, "strain_id": 2, "caseattr_id": 101, "value": "3"},
{"inbredset_id": 1, "strain_id": 3, "caseattr_id": 102, "value": "2"},
{"inbredset_id": 1, "strain_id": 4, "caseattr_id": 101, "value": "3"},
{"inbredset_id": 1, "strain_id": 5, "caseattr_id": 102, "value": "oo"}
]
)
@pytest.mark.unit_test
def test_apply_change_rejected(mocker: MockFixture) -> None:
"""Test rejecting a change"""
mock_cursor, mock_conn = mocker.MagicMock(), mocker.MagicMock()
mock_conn.cursor.return_value = mock_cursor
mock_lmdb = mocker.patch("gn3.db.case_attributes.lmdb")
mock_env, mock_txn = mocker.MagicMock(), mocker.MagicMock()
mock_lmdb.open.return_value = mock_env
mock_env.begin.return_value.__enter__.return_value = mock_txn
tmpdir = Path(os.environ.get("TMPDIR", tempfile.gettempdir()))
change_id, review_ids = 3, {1, 2, 3}
mock_txn.get.side_effect = [
pickle.dumps(review_ids), # review_ids
None # rejected_ids (initially empty)
]
assert apply_change(mock_cursor, EditStatus.rejected,
change_id, tmpdir) is True
# Verify SQL query call sequence
mock_cursor.execute.assert_called_once_with(
"UPDATE caseattributes_audit SET status = %s WHERE id = %s",
(str(EditStatus.rejected), change_id))
mock_cursor.executemany.assert_not_called()
# Verify LMDB operations
mock_env.begin.assert_called_once_with(write=True)
expected_txn_calls = [
mocker.call(b"review", pickle.dumps({1, 2})),
mocker.call(b"rejected", pickle.dumps({3}))
]
mock_txn.put.assert_has_calls(expected_txn_calls, any_order=False)
@pytest.mark.unit_test
def test_apply_change_non_existent_change_id(mocker: MockFixture) -> None:
"""Test that there's a missing change_id from the returned LMDB rejected set."""
mock_env, mock_txn = mocker.MagicMock(), mocker.MagicMock()
mock_cursor, mock_conn = mocker.MagicMock(), mocker.MagicMock()
mock_lmdb = mocker.patch("gn3.db.case_attributes.lmdb")
mock_lmdb.open.return_value = mock_env
mock_conn.cursor.return_value = mock_cursor
mock_env.begin.return_value.__enter__.return_value = mock_txn
change_id, review_ids = 28, {1, 2, 3}
mock_txn.get.side_effect = [
pickle.dumps(review_ids), # b"review" key
None, # b"approved" key
]
tmpdir = Path(os.environ.get("TMPDIR", tempfile.gettempdir()))
assert apply_change(mock_cursor, EditStatus.approved,
change_id, tmpdir) is False
|
