tests.test_root.RootTest.test_get_user_no_teams()   A
last analyzed

Complexity

Conditions 1

Size

Total Lines 10
Code Lines 4

Duplication

Lines 0
Ratio 0 %

Importance

Changes 0
Metric Value
cc 1
eloc 4
nop 1
dl 0
loc 10
rs 10
c 0
b 0
f 0
1
#!/usr/bin/env python3
2
# -*- coding: utf-8 -*-
3
"""
4
Created on Fri Jan 10 11:28:10 2020
5
6
@author: Paolo Cozzi <[email protected]>
7
"""
8
9
import os
10
import json
11
import types
12
13
from collections import defaultdict
14
from unittest.mock import patch, Mock
15
from unittest import TestCase
16
17
from pyUSIrest.auth import Auth
18
from pyUSIrest.usi import Root, Team, Submission
19
from pyUSIrest.settings import ROOT_URL
20
from pyUSIrest.exceptions import USIConnectionError, USIDataError
21
22
from .common import DATA_PATH
23
from .test_auth import generate_token
24
25
26
class RootTest(TestCase):
27
    @classmethod
28
    def setup_class(cls):
29
        cls.mock_get_patcher = patch('requests.Session.get')
30
        cls.mock_get = cls.mock_get_patcher.start()
31
32
    @classmethod
33
    def teardown_class(cls):
34
        cls.mock_get_patcher.stop()
35
36
    def setUp(self):
37
        self.auth = Auth(token=generate_token())
38
39
        with open(os.path.join(DATA_PATH, "root.json")) as handle:
40
            data = json.load(handle)
41
42
        self.mock_get.return_value = Mock()
43
        self.mock_get.return_value.json.return_value = data
44
        self.mock_get.return_value.status_code = 200
45
46
        # get a root object
47
        self.root = Root(self.auth)
48
49
    def test_str(self):
50
        test = self.root.__str__()
51
        reference = "Biosample API root at %s" % (ROOT_URL + "/api/")
52
53
        self.assertIsInstance(test, str)
54
        self.assertEqual(reference, test)
55
56
    def read_userTeams(self, filename="userTeams.json"):
57
        with open(os.path.join(DATA_PATH, filename)) as handle:
58
            data = json.load(handle)
59
60
        self.mock_get.return_value = Mock()
61
        self.mock_get.return_value.json.return_value = data
62
        self.mock_get.return_value.status_code = 200
63
64
    def test_get_user_teams(self):
65
        # initialize
66
        self.read_userTeams()
67
68
        # get user teams
69
        teams = self.root.get_user_teams()
70
71
        # teams is now a generator
72
        self.assertIsInstance(teams, types.GeneratorType)
73
        teams = list(teams)
74
75
        self.assertEqual(len(teams), 1)
76
77
        team = teams[0]
78
        self.assertIsInstance(team, Team)
79
80
    def test_get_user_no_teams(self):
81
        """Test for a user having no teams"""
82
83
        # initialize
84
        self.read_userTeams(filename="userNoTeams.json")
85
86
        # get user teams (is an empty iterator)
87
        teams = self.root.get_user_teams()
88
89
        self.assertRaises(StopIteration, next, teams)
90
91
    def test_get_team_by_name(self):
92
        # initialize
93
        self.read_userTeams()
94
95
        # get a specific team
96
        team = self.root.get_team_by_name("subs.dev-team-1")
97
        self.assertIsInstance(team, Team)
98
99
        # get a team I dont't belong to
100
        self.assertRaisesRegex(
101
            NameError,
102
            "team: .* not found",
103
            self.root.get_team_by_name,
104
            "subs.dev-team-2")
105
106
    def mocked_get_submission(*args, **kwargs):
107
        class MockResponse:
108
            def __init__(self, json_data, status_code):
109
                self.json_data = json_data
110
                self.status_code = status_code
111
                self.text = "MockResponse not implemented: %s" % (args[0])
112
113
            def json(self):
114
                return self.json_data
115
116
        # this variable will collect all replies
117
        replies = defaultdict(lambda: MockResponse(None, 404))
118
119
        # a custom function to set up replies for link
120 View Code Duplication
        def set_reply(url, filename, status=200):
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
121
            # referring to the upper replies variable
122
            nonlocal replies
0 ignored issues
show
Comprehensibility Best Practice introduced by
The variable replies does not seem to be defined.
Loading history...
123
124
            # open data file
125
            with open(os.path.join(DATA_PATH, filename)) as handle:
126
                data = json.load(handle)
127
128
            # track reply to URL
129
            replies[url] = MockResponse(data, status)
130
131
        submission_prefix = "https://submission-test.ebi.ac.uk/api/submissions"
132
        status_suffix = "submissionStatus"
133
134
        status_link1 = "/".join([
135
            submission_prefix,
136
            "87e7abda-81a8-4b5e-a1c0-323f7f0a4e43",
137
            status_suffix])
138
139
        status_link2 = "/".join([
140
            submission_prefix,
141
            "8b05e7f2-92c1-4651-94cb-9101f351f000",
142
            status_suffix])
143
144
        # --- to test get_submission_by_name
145
        byname_link = "/".join([
146
            submission_prefix, "c8c86558-8d3a-4ac5-8638-7aa354291d61"])
147
148
        status_link3 = "/".join([
149
            submission_prefix,
150
            "c8c86558-8d3a-4ac5-8638-7aa354291d61",
151
            status_suffix])
152
153
        # --- for each url, return a different response
154
        set_reply(
155
            'https://submission-test.ebi.ac.uk/api/user/submissions',
156
            "userSubmissionsPage1.json")
157
158
        set_reply(
159
            'https://submission-test.ebi.ac.uk/api/user/submissions'
160
            '?page=1&size=1',
161
            "userSubmissionsPage2.json")
162
163
        set_reply(status_link1, "submissionStatus1.json")
164
165
        set_reply(status_link2, "submissionStatus2.json")
166
167
        set_reply(byname_link, "newSubmission.json")
168
169
        set_reply(byname_link, "newSubmission.json")
170
171
        set_reply(status_link3, "submissionStatus2.json")
172
173
        return replies[args[0]]
174
175
    @patch('requests.Session.get', side_effect=mocked_get_submission)
176
    def test_get_user_submissions(self, mock_get):
177
        # get userSubmissions
178
        submissions = self.root.get_user_submissions()
179
180
        # submissions is now a generator
181
        self.assertIsInstance(submissions, types.GeneratorType)
182
183
        # convert it into a list
184
        submissions = list(submissions)
185
        self.assertEqual(len(submissions), 2)
186
187
        for submission in submissions:
188
            self.assertIsInstance(submission, Submission)
189
190
        # testing filtering
191
        draft = self.root.get_user_submissions(status="Draft")
192
193
        # submissions is now a generator
194
        self.assertIsInstance(draft, types.GeneratorType)
195
196
        # convert it into a list
197
        draft = list(draft)
198
        self.assertEqual(len(draft), 1)
199
200
        team1 = self.root.get_user_submissions(team="subs.test-team-1")
201
202
        # submissions is now a generator
203
        self.assertIsInstance(team1, types.GeneratorType)
204
205
        # convert it into a list
206
        team1 = list(team1)
207
        self.assertEqual(len(team1), 1)
208
209
        completed1 = self.root.get_user_submissions(
210
            team="subs.dev-team-1", status="Completed")
211
212
        # submissions is now a generator
213
        self.assertIsInstance(completed1, types.GeneratorType)
214
215
        # convert it into a list
216
        completed1 = list(completed1)
217
        self.assertEqual(len(completed1), 0)
218
219
    @patch('requests.Session.get', side_effect=mocked_get_submission)
220
    def test_get_submission_by_name(self, mock_get):
221
        submission = self.root.get_submission_by_name(
222
            submission_name='c8c86558-8d3a-4ac5-8638-7aa354291d61')
223
224
        self.assertIsInstance(submission, Submission)
225
226
    def test_get_submission_not_found(self):
227
        """Test get a submission with a wrong name"""
228
229
        self.mock_get.return_value = Mock()
230
        self.mock_get.return_value.json.return_value = ''
231
        self.mock_get.return_value.status_code = 404
232
233
        self.assertRaisesRegex(
234
            NameError,
235
            "submission: .* not found",
236
            self.root.get_submission_by_name,
237
            submission_name='c8c86558-8d3a-4ac5-8638-7aa354291d61')
238
239
        # a different 40x error type
240
        self.mock_get.return_value = Mock()
241
        self.mock_get.return_value.text = (
242
            "The request did not include an Authorization header")
243
        self.mock_get.return_value.status_code = 401
244
245
        self.assertRaisesRegex(
246
            USIDataError,
247
            "Error with request",
248
            self.root.get_submission_by_name,
249
            submission_name='c8c86558-8d3a-4ac5-8638-7aa354291d61')
250
251
        self.mock_get.return_value = Mock()
252
        self.mock_get.return_value.status_code = 500
253
254
        self.assertRaises(
255
            USIConnectionError,
256
            self.root.get_submission_by_name,
257
            submission_name='c8c86558-8d3a-4ac5-8638-7aa354291d61')
258