Add force_querystring argument to fetch_token() method

Author: singingwolfboy

HISTORY.rst

@@ -5,6 +5,7 @@ UNRELEASED
 ++++++++++
 
 - Instagram compliance fix
+- Added ``force_querystring`` argument to fetch_token() method on OAuth2Session
 
 v1.2.0 (14 January 2019)
 ++++++++++++++++++++++++

requests_oauthlib/oauth2_session.py

@@ -182,6 +182,7 @@ def fetch_token(
         username=None,
         password=None,
         method="POST",
+        force_querystring=False,
         timeout=None,
         headers=None,
         verify=True,
@@ -212,6 +213,8 @@ def fetch_token(
         :param method: The HTTP method used to make the request. Defaults
                        to POST, but may also be GET. Other methods should
                        be added as needed.
+        :param force_querystring: If True, force the request body to be sent
+            in the querystring instead.
         :param timeout: Timeout of the request in seconds.
         :param headers: Dict to default request headers with.
         :param verify: Verify SSL certificate.
@@ -320,33 +323,29 @@ def fetch_token(
             "Content-Type": "application/x-www-form-urlencoded;charset=UTF-8",
         }
         self.token = {}
+        request_kwargs = {}
         if method.upper() == "POST":
-            r = self.post(
-                token_url,
-                data=dict(urldecode(body)),
-                timeout=timeout,
-                headers=headers,
-                auth=auth,
-                verify=verify,
-                proxies=proxies,
+            request_kwargs["params" if force_querystring else "data"] = dict(
+                urldecode(body)
             )
-            log.debug("Prepared fetch token request body %s", body)
         elif method.upper() == "GET":
-            # if method is not 'POST', switch body to querystring and GET
-            r = self.get(
-                token_url,
-                params=dict(urldecode(body)),
-                timeout=timeout,
-                headers=headers,
-                auth=auth,
-                verify=verify,
-                proxies=proxies,
-            )
-            log.debug("Prepared fetch token request querystring %s", body)
+            request_kwargs["params"] = dict(urldecode(body))
         else:
             raise ValueError("The method kwarg must be POST or GET.")
 
+        r = self.request(
+            method=method,
+            url=token_url,
+            timeout=timeout,
+            headers=headers,
+            auth=auth,
+            verify=verify,
+            proxies=proxies,
+            **request_kwargs
+        )
+
         log.debug("Request to fetch token completed with status %s.", r.status_code)
+        log.debug("Request url was %s", r.request.url)
         log.debug("Request headers were %s", r.request.headers)
         log.debug("Request body was %s", r.request.body)
         log.debug("Response headers were %s and content %s.", r.headers, r.text)

tests/test_oauth2_session.py

@@ -25,6 +25,7 @@
 
 
 fake_time = time.time()
+CODE = "asdf345xdf"
 
 
 def fake_token(token):
@@ -51,9 +52,7 @@ def setUp(self):
         self.client_secret = "someclientsecret"
         self.user_username = "user_username"
         self.user_password = "user_password"
-        self.client_WebApplication = WebApplicationClient(
-            self.client_id, code="asdf345xdf"
-        )
+        self.client_WebApplication = WebApplicationClient(self.client_id, code=CODE)
         self.client_LegacyApplication = LegacyApplicationClient(self.client_id)
         self.client_BackendApplication = BackendApplicationClient(self.client_id)
         self.client_MobileApplication = MobileApplicationClient(self.client_id)
@@ -291,7 +290,7 @@ def fake_send(r, **kwargs):
             _fetch_history[2][2], expected_auth_header
         )  # ensure a Basic Authorization header
 
-        # scneario 4 - send in a username/password combo
+        # scenario 4 - send in a username/password combo
         # this should send the `client_id` in the headers, like scenario 1
         self.assertEqual(
             sess.fetch_token(
@@ -312,31 +311,39 @@ def fake_send(r, **kwargs):
         self.assertIn("username=%s" % self.user_username, _fetch_history[3][1])
         self.assertIn("password=%s" % self.user_password, _fetch_history[3][1])
 
+        # scenario 5 - send data in `params` and not in `data` for providers
+        # that expect data in URL
+        self.assertEqual(
+            sess.fetch_token(url, client_secret="somesecret", force_querystring=True),
+            self.token,
+        )
+        self.assertIn("code=%s" % CODE, _fetch_history[4][0])
+
         # some quick tests for valid ways of supporting `client_secret`
 
         # scenario 2b - force the `client_id` into the body; but the `client_secret` is `None`
         self.assertEqual(
             sess.fetch_token(url, client_secret=None, include_client_id=True),
             self.token,
         )
-        self.assertEqual(len(_fetch_history), 5)
-        self.assertIn("client_id=%s" % self.client_id, _fetch_history[4][1])
+        self.assertEqual(len(_fetch_history), 6)
+        self.assertIn("client_id=%s" % self.client_id, _fetch_history[5][1])
         self.assertNotIn(
-            "client_secret", _fetch_history[4][1]
+            "client_secret=", _fetch_history[5][1]
         )  # no `client_secret` in the body
         self.assertEqual(
-            _fetch_history[4][2], None
+            _fetch_history[5][2], None
         )  # ensure NO Basic Authorization header
 
         # scenario 2c - force the `client_id` into the body; but the `client_secret` is an empty string
         self.assertEqual(
             sess.fetch_token(url, client_secret="", include_client_id=True), self.token
         )
-        self.assertEqual(len(_fetch_history), 6)
-        self.assertIn("client_id=%s" % self.client_id, _fetch_history[5][1])
-        self.assertIn("client_secret=", _fetch_history[5][1])
+        self.assertEqual(len(_fetch_history), 7)
+        self.assertIn("client_id=%s" % self.client_id, _fetch_history[6][1])
+        self.assertIn("client_secret=", _fetch_history[6][1])
         self.assertEqual(
-            _fetch_history[5][2], None
+            _fetch_history[6][2], None
         )  # ensure NO Basic Authorization header
 
     def test_cleans_previous_token_before_fetching_new_one(self):