tests.test_root.RootTest.test_get_user_submissions() - Code Metrics - Inspection of ":bookmark: pyUSIrest release v0.3.0" - cnr-ibba/pyUSIrest - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#7)

by Paolo

created 2020-01-14 11:02 UTC

RootTest.test_get_user_submissions() A

↳ Parent: tests.test_root

Complexity

Conditions

Size

Total Lines	43
Code Lines	21

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	2
eloc	21
nop	2
dl	0
loc	43
rs	9.376
c	0
b	0
f	0

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Created on Fri Jan 10 11:28:10 2020

@author: Paolo Cozzi <[email protected]>
"""

import os
import json
import types

from collections import defaultdict
from unittest.mock import patch, Mock
from unittest import TestCase

from pyUSIrest.auth import Auth
from pyUSIrest.usi import Root, Team, Submission
from pyUSIrest.settings import ROOT_URL

from .common import DATA_PATH
from .test_auth import generate_token


class RootTest(TestCase):
    @classmethod
    def setup_class(cls):
        cls.mock_get_patcher = patch('requests.Session.get')
        cls.mock_get = cls.mock_get_patcher.start()

    @classmethod
    def teardown_class(cls):
        cls.mock_get_patcher.stop()

    def setUp(self):
        self.auth = Auth(token=generate_token())

        with open(os.path.join(DATA_PATH, "root.json")) as handle:
            data = json.load(handle)

        self.mock_get.return_value = Mock()
        self.mock_get.return_value.json.return_value = data
        self.mock_get.return_value.status_code = 200

        # get a root object
        self.root = Root(self.auth)

    def test_str(self):
        test = self.root.__str__()
        reference = "Biosample API root at %s" % (ROOT_URL + "/api/")

        self.assertIsInstance(test, str)
        self.assertEqual(reference, test)

    def read_userTeams(self, filename="userTeams.json"):
        with open(os.path.join(DATA_PATH, filename)) as handle:
            data = json.load(handle)

        self.mock_get.return_value = Mock()
        self.mock_get.return_value.json.return_value = data
        self.mock_get.return_value.status_code = 200

    def test_get_user_teams(self):
        # initialize
        self.read_userTeams()

        # get user teams
        teams = self.root.get_user_teams()

        # teams is now a generator
        self.assertIsInstance(teams, types.GeneratorType)
        teams = list(teams)

        self.assertEqual(len(teams), 1)

        team = teams[0]
        self.assertIsInstance(team, Team)

    def test_get_user_no_teams(self):
        """Test for a user having no teams"""

        # initialize
        self.read_userTeams(filename="userNoTeams.json")

        # get user teams (is an empty iterator)
        teams = self.root.get_user_teams()

        self.assertRaises(StopIteration, next, teams)

    def test_get_team_by_name(self):
        # initialize
        self.read_userTeams()

        # get a specific team
        team = self.root.get_team_by_name("subs.dev-team-1")
        self.assertIsInstance(team, Team)

        # get a team I dont't belong to
        self.assertRaisesRegex(
            NameError,
            "team: .* not found",
            self.root.get_team_by_name,
            "subs.dev-team-2")

    def mocked_get_submission(*args, **kwargs):
        class MockResponse:
            def __init__(self, json_data, status_code):
                self.json_data = json_data
                self.status_code = status_code
                self.text = "MockResponse not implemented: %s" % (args[0])

            def json(self):
                return self.json_data

        # this variable will collect all replies
        replies = defaultdict(lambda: MockResponse(None, 404))

        # a custom function to set up replies for link
        def set_reply(url, filename, status=200):

            # referring to the upper replies variable
            nonlocal replies


            # open data file
            with open(os.path.join(DATA_PATH, filename)) as handle:
                data = json.load(handle)

            # track reply to URL
            replies[url] = MockResponse(data, status)

        submission_prefix = "https://submission-test.ebi.ac.uk/api/submissions"
        status_suffix = "submissionStatus"

        status_link1 = "/".join([
            submission_prefix,
            "87e7abda-81a8-4b5e-a1c0-323f7f0a4e43",
            status_suffix])

        status_link2 = "/".join([
            submission_prefix,
            "8b05e7f2-92c1-4651-94cb-9101f351f000",
            status_suffix])

        # --- to test get_submission_by_name
        byname_link = "/".join([
            submission_prefix, "c8c86558-8d3a-4ac5-8638-7aa354291d61"])

        status_link3 = "/".join([
            submission_prefix,
            "c8c86558-8d3a-4ac5-8638-7aa354291d61",
            status_suffix])

        # --- for each url, return a different response
        set_reply(
            'https://submission-test.ebi.ac.uk/api/user/submissions',
            "userSubmissionsPage1.json")

        set_reply(
            'https://submission-test.ebi.ac.uk/api/user/submissions'
            '?page=1&size=1',
            "userSubmissionsPage2.json")

        set_reply(status_link1, "submissionStatus1.json")

        set_reply(status_link2, "submissionStatus2.json")

        set_reply(byname_link, "newSubmission.json")

        set_reply(byname_link, "newSubmission.json")

        set_reply(status_link3, "submissionStatus2.json")

        return replies[args[0]]

    @patch('requests.Session.get', side_effect=mocked_get_submission)
    def test_get_user_submissions(self, mock_get):
        # get userSubmissions
        submissions = self.root.get_user_submissions()

        # submissions is now a generator
        self.assertIsInstance(submissions, types.GeneratorType)

        # convert it into a list
        submissions = list(submissions)
        self.assertEqual(len(submissions), 2)

        for submission in submissions:
            self.assertIsInstance(submission, Submission)

        # testing filtering
        draft = self.root.get_user_submissions(status="Draft")

        # submissions is now a generator
        self.assertIsInstance(draft, types.GeneratorType)

        # convert it into a list
        draft = list(draft)
        self.assertEqual(len(draft), 1)

        team1 = self.root.get_user_submissions(team="subs.test-team-1")

        # submissions is now a generator
        self.assertIsInstance(team1, types.GeneratorType)

        # convert it into a list
        team1 = list(team1)
        self.assertEqual(len(team1), 1)

        completed1 = self.root.get_user_submissions(
            team="subs.dev-team-1", status="Completed")

        # submissions is now a generator
        self.assertIsInstance(completed1, types.GeneratorType)

        # convert it into a list
        completed1 = list(completed1)
        self.assertEqual(len(completed1), 0)

    @patch('requests.Session.get', side_effect=mocked_get_submission)
    def test_get_submission_by_name(self, mock_get):
        submission = self.root.get_submission_by_name(
            submission_name='c8c86558-8d3a-4ac5-8638-7aa354291d61')

        self.assertIsInstance(submission, Submission)

    def test_get_submission_not_found(self):
        """Test get a submission with a wrong name"""

        self.mock_get.return_value = Mock()
        self.mock_get.return_value.json.return_value = ''
        self.mock_get.return_value.status_code = 404

        self.assertRaisesRegex(
            NameError,
            "submission: .* not found",
            self.root.get_submission_by_name,
            submission_name='c8c86558-8d3a-4ac5-8638-7aa354291d61')

        self.mock_get.return_value = Mock()
        self.mock_get.return_value.status_code = 500

        self.assertRaises(
            ConnectionError,
            self.root.get_submission_by_name,
            submission_name='c8c86558-8d3a-4ac5-8638-7aa354291d61')


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

cnr-ibba / pyUSIrest

Pull Request — master (#7)

RootTest.test_get_user_submissions() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like