Users should be able to override the token when using the Slack compl…

…iance fix

requests_oauthlib/compliance_fixes/slack.py

@@ -1,10 +1,36 @@
+try:
+    from urlparse import urlparse, parse_qs
+except ImportError:
+    from urllib.parse import urlparse, parse_qs
+
 from oauthlib.common import add_params_to_uri
 
 
 def slack_compliance_fix(session):
     def _non_compliant_param_name(url, headers, data):
-        token = [('token', session.access_token)]
-        url = add_params_to_uri(url, token)
+        # If the user has already specified the token, either in the URL
+        # or in a data dictionary, then there's nothing to do.
+        # If the specified token is different from ``session.access_token``,
+        # we assume the user intends to override the access token.
+        url_query = dict(parse_qs(urlparse(url).query))
+        token = url_query.get("token")
+        if not token and isinstance(data, dict):
+            token = data.get("token")
+
+        if token:
+            # Nothing to do, just return.
+            return url, headers, data
+
+        if not data:
+            data = {"token": session.access_token}
+        elif isinstance(data, dict):
+            data["token"] = session.access_token
+        else:
+            # ``data`` is something other than a dict: maybe a stream,
+            # maybe a file object, maybe something else. We can't easily
+            # modify it, so we'll set the token by modifying the URL instead.
+            token = [('token', session.access_token)]
+            url = add_params_to_uri(url, token)
         return url, headers, data
 
     session.register_compliance_hook('protected_request', _non_compliant_param_name)

tests/test_compliance_fixes.py

@@ -150,17 +150,19 @@ def setUp(self):
               "scope": "read",
             },
         )
-        mocker.get(
-            "https://slack.com/api/auth.test",
-            json={
-              "ok": True,
-              "url": "https://myteam.slack.com/",
-              "team": "My Team",
-              "user": "cal",
-              "team_id": "T12345",
-              "user_id": "U12345",
-            }
-        )
+        for method in ("GET", "POST"):
+            mocker.request(
+                method=method,
+                url="https://slack.com/api/auth.test",
+                json={
+                  "ok": True,
+                  "url": "https://myteam.slack.com/",
+                  "team": "My Team",
+                  "user": "cal",
+                  "team_id": "T12345",
+                  "user_id": "U12345",
+                }
+            )
         mocker.start()
         self.addCleanup(mocker.stop)
 
@@ -174,4 +176,43 @@ def test_protected_request(self):
         )
         url = response.request.url
         query = parse_qs(urlparse(url).query)
-        self.assertEqual(query["token"], ["dummy-access-token"])
+        self.assertNotIn("token", query)
+        body = response.request.body
+        data = parse_qs(body)
+        self.assertEqual(data["token"], ["dummy-access-token"])
+
+    def test_protected_request_override_token_get(self):
+        self.session.token = {"access_token": 'dummy-access-token'}
+        response = self.session.get(
+            "https://slack.com/api/auth.test",
+            data={"token": "different-token"},
+        )
+        url = response.request.url
+        query = parse_qs(urlparse(url).query)
+        self.assertNotIn("token", query)
+        body = response.request.body
+        data = parse_qs(body)
+        self.assertEqual(data["token"], ["different-token"])
+
+    def test_protected_request_override_token_post(self):
+        self.session.token = {"access_token": 'dummy-access-token'}
+        response = self.session.post(
+            "https://slack.com/api/auth.test",
+            data={"token": "different-token"},
+        )
+        url = response.request.url
+        query = parse_qs(urlparse(url).query)
+        self.assertNotIn("token", query)
+        body = response.request.body
+        data = parse_qs(body)
+        self.assertEqual(data["token"], ["different-token"])
+
+    def test_protected_request_override_token_url(self):
+        self.session.token = {"access_token": 'dummy-access-token'}
+        response = self.session.get(
+            "https://slack.com/api/auth.test?token=different-token",
+        )
+        url = response.request.url
+        query = parse_qs(urlparse(url).query)
+        self.assertEqual(query["token"], ["different-token"])
+        self.assertIsNone(response.request.body)