tests.test_root.RootTest.mocked_get_submission() - Code Metrics - Inspection of ":recycle: Use iterators when browsing USI objects" - cnr-ibba/pyUSIrest - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — devel (#6)

by Paolo

created 2020-01-10 10:38 UTC

tests.test_root.RootTest.mocked_get_submission() B

↳ Parent: tests.test_root

Complexity

Conditions

Size

Total Lines	81
Code Lines	50

Duplication

Lines	10
Ratio	12.35 %

Importance

Changes

Metric	Value
cc	3
eloc	50
nop	2
dl	10
loc	81
rs	8.6363
c	0
b	0
f	0

How to fix Long Method

#!/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")

        # to reload submissions
        set_reply(
            submission_prefix + "/" +
            "87e7abda-81a8-4b5e-a1c0-323f7f0a4e43",
            "Submission1.json"
            )

        set_reply(
            submission_prefix + "/" +
            "8b05e7f2-92c1-4651-94cb-9101f351f000",
            "Submission2.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		# to reload submissions
173		set_reply(
174		submission_prefix + "/" +
175		"87e7abda-81a8-4b5e-a1c0-323f7f0a4e43",
176		"Submission1.json"
177		)
178
179		set_reply(
180		submission_prefix + "/" +
181		"8b05e7f2-92c1-4651-94cb-9101f351f000",
182		"Submission2.json"
183		)
184
185		return replies[args[0]]
186
187		@patch('requests.Session.get', side_effect=mocked_get_submission)
188		def test_get_user_submissions(self, mock_get):
189		# get userSubmissions
190		submissions = self.root.get_user_submissions()
191
192		# submissions is now a generator
193		self.assertIsInstance(submissions, types.GeneratorType)
194
195		# convert it into a list
196		submissions = list(submissions)
197		self.assertEqual(len(submissions), 2)
198
199		for submission in submissions:
200		self.assertIsInstance(submission, Submission)
201
202		# testing filtering
203		draft = self.root.get_user_submissions(status="Draft")
204
205		# submissions is now a generator
206		self.assertIsInstance(draft, types.GeneratorType)
207
208		# convert it into a list
209		draft = list(draft)
210		self.assertEqual(len(draft), 1)
211
212		team1 = self.root.get_user_submissions(team="subs.test-team-1")
213
214		# submissions is now a generator
215		self.assertIsInstance(team1, types.GeneratorType)
216
217		# convert it into a list
218		team1 = list(team1)
219		self.assertEqual(len(team1), 1)
220
221		completed1 = self.root.get_user_submissions(
222		team="subs.dev-team-1", status="Completed")
223
224		# submissions is now a generator
225		self.assertIsInstance(completed1, types.GeneratorType)
226
227		# convert it into a list
228		completed1 = list(completed1)
229		self.assertEqual(len(completed1), 0)
230
231		@patch('requests.Session.get', side_effect=mocked_get_submission)
232		def test_get_submission_by_name(self, mock_get):
233		submission = self.root.get_submission_by_name(
234		submission_name='c8c86558-8d3a-4ac5-8638-7aa354291d61')
235
236		self.assertIsInstance(submission, Submission)
237
238		def test_get_submission_not_found(self):
239		"""Test get a submission with a wrong name"""
240
241		self.mock_get.return_value = Mock()
242		self.mock_get.return_value.json.return_value = ''
243		self.mock_get.return_value.status_code = 404
244
245		self.assertRaisesRegex(
246		NameError,
247		"submission: .* not found",
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		ConnectionError,
256		self.root.get_submission_by_name,
257		submission_name='c8c86558-8d3a-4ac5-8638-7aa354291d61')
258

cnr-ibba / pyUSIrest

Pull Request — devel (#6)

tests.test_root.RootTest.mocked_get_submission() B

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like