Unit tests for set_credentials in tabpy-tools.

.gitignore

@@ -47,6 +47,7 @@ coverage.xml
 *.cover
 .hypothesis/
 .pytest_cache/
+*_trial_temp*
 
 # Translations
 *.mo

tabpy-tools/tabpy_tools/rest.py

@@ -3,11 +3,10 @@
 import requests
 from requests.auth import HTTPBasicAuth
 from re import compile
+import json
 
 from collections import MutableMapping as _MutableMapping
 
-import json as json
-
 
 logger = logging.getLogger(__name__)
 
@@ -25,7 +24,6 @@ def __init__(self, response):
             self.info = r['info']
             self.message = response.json()['message']
         except (json.JSONDecodeError,
-                simplejson.errors.JSONDecodeError,
                 KeyError):
             self.info = None
             self.message = response.text
@@ -59,14 +57,16 @@ def __init__(self, session=None):
         self.session = session
         self.auth = None
 
-    def raise_error(self, response):
+    @staticmethod
+    def raise_error(response):
         logger.error("Error with server response. code=%s; text=%s",
                      response.status_code,
                      response.text)
 
         raise ResponseError(response)
 
-    def _remove_nones(self, data):
+    @staticmethod
+    def _remove_nones(data):
         if isinstance(data, dict):
             for k in [k for k, v in data.items() if v is None]:
                 del data[k]
@@ -187,6 +187,7 @@ class ServiceClient(object):
     endpoint URL is prepended to all the requests and forwarded to the network
     wrapper.
     """
+
     def __init__(self, endpoint, network_wrapper=None):
         if network_wrapper is None:
             network_wrapper = RequestsNetworkWrapper(
@@ -325,12 +326,12 @@ def __init__(self, **kwargs):
 
     def __repr__(self):
         return (
-                "{" +
-                ", ".join([
-                    repr(k) + ": " + repr(v)
-                    for k, v in self.items()
-                ]) +
-                "}"
+            "{" +
+            ", ".join([
+                repr(k) + ": " + repr(v)
+                for k, v in self.items()
+            ]) +
+            "}"
         )
 
     @classmethod
@@ -365,7 +366,7 @@ def to_json(self):
         return result
 
     def __eq__(self, other):
-        return (type(self) == type(other) and
+        return (isinstance(self, type(other)) and
                 all((
                     getattr(self, a) == getattr(other, a)
                     for a in self.__rest__

tabpy-tools/tools_tests/test_client.py

@@ -1,8 +1,6 @@
 import unittest
 from unittest.mock import Mock
 
-import requests
-
 from tabpy_tools.client import Client
 
 
@@ -56,7 +54,7 @@ def test_query(self):
 
         self.assertEqual(self.client.query("foo", 1, 2, 3), "ok")
 
-        self.client._service.query.sssert_called_once_with("foo", 1, 2, 3)
+        self.client._service.query.assert_called_once_with("foo", 1, 2, 3)
 
         self.client._service.query.reset_mock()
 
@@ -77,3 +75,10 @@ def test_get_endpoint_upload_destination(self):
             {"path": "foo"}
 
         self.assertEqual(self.client._get_endpoint_upload_destination(), "foo")
+
+    def test_set_credentials(self):
+        username, password = "username", "password"
+        self.client.set_credentials(username, password)
+
+        self.client._service.set_credentials.assert_called_once_with(
+            username, password)

tabpy-tools/tools_tests/test_rest.py

@@ -3,6 +3,7 @@
 import sys
 import unittest
 from unittest.mock import Mock
+from requests.auth import HTTPBasicAuth
 
 from tabpy_tools.rest import (RequestsNetworkWrapper, ServiceClient)
 
@@ -148,6 +149,69 @@ def test_DELETE_InvalidURL(self):
             e = sys.exc_info()[0]
             self.assertEqual(e, TypeError)
 
+    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):
 
@@ -179,3 +243,9 @@ 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_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)