tests.unit.tools_tests.test_rest.TestRequestsNetworkWrapper.test_GET() - Code Metrics - tableau/TabPy - Measure and Improve Code Quality continuously with Scrutinizer

TestRequestsNetworkWrapper.test_GET() A
last analyzed 2024-11-25 22:00 UTC

↳ Parent: tests.unit.tools_tests.test_rest

Complexity

Conditions

Size

Total Lines	6
Code Lines	6

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	6
dl	0
loc	6
rs	10
c	0
b	0
f	0
cc	1
nop	1

import json
import requests
from requests.auth import HTTPBasicAuth
from tabpy.tabpy_tools.rest import RequestsNetworkWrapper, ServiceClient
import unittest
from unittest.mock import Mock


class TestRequestsNetworkWrapper(unittest.TestCase):
    def test_init(self):
        RequestsNetworkWrapper()

    def test_init_with_session(self):
        session = {}

        rnw = RequestsNetworkWrapper(session=session)

        self.assertIs(session, rnw.session)

    def mock_response(self, status_code):
        response = Mock(requests.Response)
        response.json.return_value = "json"
        response.status_code = status_code

        return response

    def setUp(self):
        session = Mock(requests.Session)
        session.get.return_value = self.mock_response(200)
        session.post.return_value = self.mock_response(200)
        session.put.return_value = self.mock_response(200)
        session.delete.return_value = self.mock_response(204)

        self.rnw = RequestsNetworkWrapper(session=session)

    def test_GET(self):
        url = "abc"
        data = {"foo": "bar"}
        self.assertEqual(self.rnw.GET(url, data), "json")
        self.rnw.session.get.assert_called_once_with(
            url, params=data, timeout=None, auth=None
        )

    def test_GET_InvalidData(self):
        url = "abc"
        data = {"cat"}
        with self.assertRaises(TypeError):
            self.rnw.session.get.return_value = self.mock_response(404)
            self.rnw.GET(url, data)

    def test_GET_InvalidURL(self):
        url = ""
        data = {"foo": "bar"}
        with self.assertRaises(TypeError):
            self.rnw.session.get.return_value = self.mock_response(404)
            self.rnw.GET(url, data)

    def test_POST(self):
        url = "abc"
        data = {"foo": "bar"}
        self.assertEqual(self.rnw.POST(url, data), "json")
        self.rnw.session.post.assert_called_once_with(
            url,
            data=json.dumps(data),
            headers={"content-type": "application/json"},
            timeout=None,
            auth=None,
        )

    def test_POST_InvalidURL(self):
        url = ""
        data = {"foo": "bar"}
        with self.assertRaises(TypeError):
            self.rnw.session.post.return_value = self.mock_response(404)
            self.rnw.POST(url, data)

    def test_POST_InvalidData(self):
        url = "url"
        data = {"cat"}
        with self.assertRaises(TypeError):
            self.rnw.POST(url, data)

    def test_PUT(self):
        url = "abc"
        data = {"foo": "bar"}
        self.assertEqual(self.rnw.PUT(url, data), "json")
        self.rnw.session.put.assert_called_once_with(
            url,
            data=json.dumps(data),
            headers={"content-type": "application/json"},
            timeout=None,
            auth=None,
        )

    def test_PUT_InvalidData(self):
        url = "url"
        data = {"cat"}
        with self.assertRaises(TypeError):
            self.rnw.PUT(url, data)

    def test_PUT_InvalidURL(self):
        url = ""
        data = {"foo:bar"}
        with self.assertRaises(TypeError):
            self.rnw.PUT(url, data)

    def test_DELETE(self):
        url = "abc"
        data = {"foo": "bar"}
        self.assertIs(self.rnw.DELETE(url, data), None)
        self.rnw.session.delete.assert_called_once_with(
            url, data=json.dumps(data), timeout=None, auth=None
        )

    def test_DELETE_InvalidData(self):
        url = "abc"
        data = {"cat"}
        with self.assertRaises(TypeError):
            self.rnw.DELETE(url, data)

    def test_DELETE_InvalidURL(self):
        url = ""
        data = {"foo:bar"}
        with self.assertRaises(TypeError):
            self.rnw.DELETE(url, data)

    def test_set_credentials(self):
        expected_auth = None
        self.assertEqual(self.rnw.auth, expected_auth)

        username, password = "username", "password"
        expected_auth = HTTPBasicAuth(username, password)
        self.rnw.set_credentials(username, password)
        self.assertEqual(self.rnw.auth, expected_auth)

    def _test_METHOD_with_credentials(
        self,
        http_method_function,
        http_session_method_function,
        headers=None,
        params=False,
        data=False,
        response=None,
    ):
        username, password = "username", "password"
        self.rnw.set_credentials(username, password)

        url = "url"
        _data = {"foo": "bar"}

        self.assertEqual(http_method_function(url, _data), response)

        pargs = {url}
        kwargs = {"timeout": None, "auth": self.rnw.auth}
        if data:
            kwargs["data"] = json.dumps(_data)
        if headers:
            kwargs["headers"] = headers
        if params:
            kwargs["params"] = _data

        http_session_method_function.assert_called_once_with(*pargs, **kwargs)
        self.assertEqual(self.rnw.auth, HTTPBasicAuth(username, password))

    def test_GET_with_credentials(self):
        self._test_METHOD_with_credentials(
            self.rnw.GET, self.rnw.session.get, params=True, response="json"
        )

    def test_POST_with_credentials(self):
        self._test_METHOD_with_credentials(
            self.rnw.POST,
            self.rnw.session.post,
            headers={"content-type": "application/json"},
            data=True,
            response="json",
        )

    def test_PUT_with_credentials(self):
        self._test_METHOD_with_credentials(
            self.rnw.PUT,
            self.rnw.session.put,
            data=True,
            headers={"content-type": "application/json"},
            response="json",
        )

    def test_DELETE_with_credentials(self):
        self._test_METHOD_with_credentials(
            self.rnw.DELETE, self.rnw.session.delete, data=True
        )


class TestServiceClient(unittest.TestCase):
    def setUp(self):
        nw = Mock(RequestsNetworkWrapper())
        nw.GET.return_value = "GET"
        nw.POST.return_value = "POST"
        nw.PUT.return_value = "PUT"
        nw.DELETE.return_value = "DELETE"

        self.sc = ServiceClient("endpoint/", network_wrapper=nw)
        self.scClientDoesNotEndWithSlash = ServiceClient("endpoint", network_wrapper=nw)

    def test_GET(self):
        self.assertEqual(self.sc.GET("test"), "GET")
        self.sc.network_wrapper.GET.assert_called_once_with("endpoint/test", None, None)

    def test_POST(self):
        self.assertEqual(self.sc.POST("test"), "POST")
        self.sc.network_wrapper.POST.assert_called_once_with(
            "endpoint/test", None, None
        )

    def test_PUT(self):
        self.assertEqual(self.sc.PUT("test"), "PUT")
        self.sc.network_wrapper.PUT.assert_called_once_with("endpoint/test", None, None)

    def test_DELETE(self):
        self.assertEqual(self.sc.DELETE("test"), None)
        self.sc.network_wrapper.DELETE.assert_called_once_with(
            "endpoint/test", None, None
        )

    def test_FixEndpoint(self):
        self.assertEqual(self.scClientDoesNotEndWithSlash.GET("test"), "GET")
        self.sc.network_wrapper.GET.assert_called_once_with("endpoint/test", None, None)

    def test_set_credentials(self):
        username, password = "username", "password"
        self.sc.set_credentials(username, password)
        self.sc.network_wrapper.set_credentials.assert_called_once_with(
            username, password
        )


1			import json
2			import requests
3			from requests.auth import HTTPBasicAuth
4			from tabpy.tabpy_tools.rest import RequestsNetworkWrapper, ServiceClient
5			import unittest
6			from unittest.mock import Mock
7
8
9			class TestRequestsNetworkWrapper(unittest.TestCase):
10			def test_init(self):
11			RequestsNetworkWrapper()
12
13			def test_init_with_session(self):
14			session = {}
15
16			rnw = RequestsNetworkWrapper(session=session)
17
18			self.assertIs(session, rnw.session)
19
20			def mock_response(self, status_code):
21			response = Mock(requests.Response)
22			response.json.return_value = "json"
23			response.status_code = status_code
24
25			return response
26
27			def setUp(self):
28			session = Mock(requests.Session)
29			session.get.return_value = self.mock_response(200)
30			session.post.return_value = self.mock_response(200)
31			session.put.return_value = self.mock_response(200)
32			session.delete.return_value = self.mock_response(204)
33
34			self.rnw = RequestsNetworkWrapper(session=session)
35
36			def test_GET(self):
37			url = "abc"
38			data = {"foo": "bar"}
39			self.assertEqual(self.rnw.GET(url, data), "json")
40			self.rnw.session.get.assert_called_once_with(
41			url, params=data, timeout=None, auth=None
42			)
43
44			def test_GET_InvalidData(self):
45			url = "abc"
46			data = {"cat"}
47			with self.assertRaises(TypeError):
48			self.rnw.session.get.return_value = self.mock_response(404)
49			self.rnw.GET(url, data)
50
51			def test_GET_InvalidURL(self):
52			url = ""
53			data = {"foo": "bar"}
54			with self.assertRaises(TypeError):
55			self.rnw.session.get.return_value = self.mock_response(404)
56			self.rnw.GET(url, data)
57
58			def test_POST(self):
59			url = "abc"
60			data = {"foo": "bar"}
61			self.assertEqual(self.rnw.POST(url, data), "json")
62			self.rnw.session.post.assert_called_once_with(
63			url,
64			data=json.dumps(data),
65			headers={"content-type": "application/json"},
66			timeout=None,
67			auth=None,
68			)
69
70			def test_POST_InvalidURL(self):
71			url = ""
72			data = {"foo": "bar"}
73			with self.assertRaises(TypeError):
74			self.rnw.session.post.return_value = self.mock_response(404)
75			self.rnw.POST(url, data)
76
77			def test_POST_InvalidData(self):
78			url = "url"
79			data = {"cat"}
80			with self.assertRaises(TypeError):
81			self.rnw.POST(url, data)
82
83			def test_PUT(self):
84			url = "abc"
85			data = {"foo": "bar"}
86			self.assertEqual(self.rnw.PUT(url, data), "json")
87			self.rnw.session.put.assert_called_once_with(
88			url,
89			data=json.dumps(data),
90			headers={"content-type": "application/json"},
91			timeout=None,
92			auth=None,
93			)
94
95			def test_PUT_InvalidData(self):
96			url = "url"
97			data = {"cat"}
98			with self.assertRaises(TypeError):
99			self.rnw.PUT(url, data)
100
101			def test_PUT_InvalidURL(self):
102			url = ""
103			data = {"foo:bar"}
104			with self.assertRaises(TypeError):
105			self.rnw.PUT(url, data)
106
107			def test_DELETE(self):
108			url = "abc"
109			data = {"foo": "bar"}
110			self.assertIs(self.rnw.DELETE(url, data), None)
111			self.rnw.session.delete.assert_called_once_with(
112			url, data=json.dumps(data), timeout=None, auth=None
113			)
114
115			def test_DELETE_InvalidData(self):
116			url = "abc"
117			data = {"cat"}
118			with self.assertRaises(TypeError):
119			self.rnw.DELETE(url, data)
120
121			def test_DELETE_InvalidURL(self):
122			url = ""
123			data = {"foo:bar"}
124			with self.assertRaises(TypeError):
125			self.rnw.DELETE(url, data)
126
127			def test_set_credentials(self):
128			expected_auth = None
129			self.assertEqual(self.rnw.auth, expected_auth)
130
131			username, password = "username", "password"
132			expected_auth = HTTPBasicAuth(username, password)
133			self.rnw.set_credentials(username, password)
134			self.assertEqual(self.rnw.auth, expected_auth)
135
136			def _test_METHOD_with_credentials(
137			self,
138			http_method_function,
139			http_session_method_function,
140			headers=None,
141			params=False,
142			data=False,
143			response=None,
144			):
145			username, password = "username", "password"
146			self.rnw.set_credentials(username, password)
147
148			url = "url"
149			_data = {"foo": "bar"}
150
151			self.assertEqual(http_method_function(url, _data), response)
152
153			pargs = {url}
154			kwargs = {"timeout": None, "auth": self.rnw.auth}
155			if data:
156			kwargs["data"] = json.dumps(_data)
157			if headers:
158			kwargs["headers"] = headers
159			if params:
160			kwargs["params"] = _data
161
162			http_session_method_function.assert_called_once_with(pargs, *kwargs)
163			self.assertEqual(self.rnw.auth, HTTPBasicAuth(username, password))
164
165			def test_GET_with_credentials(self):
166			self._test_METHOD_with_credentials(
167			self.rnw.GET, self.rnw.session.get, params=True, response="json"
168			)
169
170			def test_POST_with_credentials(self):
171			self._test_METHOD_with_credentials(
172			self.rnw.POST,
173			self.rnw.session.post,
174			headers={"content-type": "application/json"},
175			data=True,
176			response="json",
177			)
178
179			def test_PUT_with_credentials(self):
180			self._test_METHOD_with_credentials(
181			self.rnw.PUT,
182			self.rnw.session.put,
183			data=True,
184			headers={"content-type": "application/json"},
185			response="json",
186			)
187
188			def test_DELETE_with_credentials(self):
189			self._test_METHOD_with_credentials(
190			self.rnw.DELETE, self.rnw.session.delete, data=True
191			)
192
193
194			class TestServiceClient(unittest.TestCase):
195			def setUp(self):
196			nw = Mock(RequestsNetworkWrapper())
197			nw.GET.return_value = "GET"
198			nw.POST.return_value = "POST"
199			nw.PUT.return_value = "PUT"
200			nw.DELETE.return_value = "DELETE"
201
202			self.sc = ServiceClient("endpoint/", network_wrapper=nw)
203			self.scClientDoesNotEndWithSlash = ServiceClient("endpoint", network_wrapper=nw)
204
205			def test_GET(self):
206			self.assertEqual(self.sc.GET("test"), "GET")
207			self.sc.network_wrapper.GET.assert_called_once_with("endpoint/test", None, None)
208
209			def test_POST(self):
210			self.assertEqual(self.sc.POST("test"), "POST")
211			self.sc.network_wrapper.POST.assert_called_once_with(
212			"endpoint/test", None, None
213			)
214
215			def test_PUT(self):
216			self.assertEqual(self.sc.PUT("test"), "PUT")
217			self.sc.network_wrapper.PUT.assert_called_once_with("endpoint/test", None, None)
218
219			def test_DELETE(self):
220			self.assertEqual(self.sc.DELETE("test"), None)
221			self.sc.network_wrapper.DELETE.assert_called_once_with(
222			"endpoint/test", None, None
223			)
224
225			def test_FixEndpoint(self):
226			self.assertEqual(self.scClientDoesNotEndWithSlash.GET("test"), "GET")
227			self.sc.network_wrapper.GET.assert_called_once_with("endpoint/test", None, None)
228
229			def test_set_credentials(self):
230			username, password = "username", "password"
231			self.sc.set_credentials(username, password)
232			self.sc.network_wrapper.set_credentials.assert_called_once_with(
233			username, password
234			)
235

tableau / TabPy

TestRequestsNetworkWrapper.test_GET() A last analyzed 2024-11-25 22:00 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

TestRequestsNetworkWrapper.test_GET() A
last analyzed 2024-11-25 22:00 UTC