Merge 7379f2c into c8a52f4

.gitignore

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

tabpy-tools/tabpy_tools/rest.py

@@ -3,11 +3,11 @@
 import requests
 from requests.auth import HTTPBasicAuth
 from re import compile
+import json
+import simplejson
 
 from collections import MutableMapping as _MutableMapping
 
-import json as json
-
 
 logger = logging.getLogger(__name__)
 

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)