Merge pull request #215 from singingwolfboy/slack-compliance-tests

Use requests-mock library, add tests for Slack compliance fix

.travis.yml

@@ -9,6 +9,7 @@ python:
   - 3.5
 install:
   - pip install -r requirements.txt
+  - pip install requests-mock
   - pip install coveralls
   - if [[ $TRAVIS_PYTHON_VERSION == "2.6" ]]; then pip install unittest2; fi
 script:

requests_oauthlib/compliance_fixes/linkedin.py

@@ -12,7 +12,7 @@ def _missing_token_type(r):
         return r
 
     def _non_compliant_param_name(url, headers, data):
-        token = [('oauth2_access_token', session._client.access_token)]
+        token = [('oauth2_access_token', session.access_token)]
         url = add_params_to_uri(url, token)
         return url, headers, data
 

requests_oauthlib/compliance_fixes/slack.py

@@ -3,7 +3,7 @@
 
 def slack_compliance_fix(session):
     def _non_compliant_param_name(url, headers, data):
-        token = [('token', session._client.access_token)]
+        token = [('token', session.access_token)]
         url = add_params_to_uri(url, token)
         return url, headers, data
 

setup.py

@@ -32,7 +32,7 @@
     os.system('python setup.py sdist upload')
     sys.exit()
 
-tests_require = ['mock']
+tests_require = ['mock', 'requests-mock']
 if sys.version_info < (2, 7): # Python 2.6 or lower
     tests_require.append('unittest2')
 

tests/test_compliance_fixes.py

@@ -1,72 +1,109 @@
 from __future__ import unicode_literals
-import unittest
+try:
+    from unittest2 import TestCase
+except ImportError:
+    from unittest import TestCase
 
-import mock
 import requests
+import requests_mock
 import time
+try:
+    from urlparse import urlparse, parse_qs
+except ImportError:
+    from urllib.parse import urlparse, parse_qs
 
 from requests_oauthlib import OAuth2Session
 from requests_oauthlib.compliance_fixes import facebook_compliance_fix
 from requests_oauthlib.compliance_fixes import linkedin_compliance_fix
 from requests_oauthlib.compliance_fixes import mailchimp_compliance_fix
 from requests_oauthlib.compliance_fixes import weibo_compliance_fix
+from requests_oauthlib.compliance_fixes import slack_compliance_fix
 
 
-class FacebookComplianceFixTest(unittest.TestCase):
+class FacebookComplianceFixTest(TestCase):
+
+    def setUp(self):
+        mocker = requests_mock.Mocker()
+        mocker.post(
+            "https://graph.facebook.com/oauth/access_token",
+            text="access_token=urlencoded",
+            headers={"Content-Type": "text/plain"},
+        )
+        mocker.start()
+        self.addCleanup(mocker.stop)
 
-    def test_fetch_access_token(self):
         facebook = OAuth2Session('foo', redirect_uri='https://i.b')
-        facebook = facebook_compliance_fix(facebook)
-
-        facebook.post = mock.MagicMock()
-        response = requests.Response()
-        response.status_code = 200
-        response.request = mock.MagicMock()
-        response._content = 'access_token=urlencoded'.encode('UTF-8')
-        response.headers['Content-Type'] = 'text/plain'
-        facebook.post.return_value = response
-
-        token = facebook.fetch_token('https://mocked.out',
-                                     client_secret='bar',
-                                     authorization_response='https://i.b/?code=hello')
+        self.session = facebook_compliance_fix(facebook)
+
+    def test_fetch_access_token(self):
+        token = self.session.fetch_token(
+            'https://graph.facebook.com/oauth/access_token',
+             client_secret='bar',
+             authorization_response='https://i.b/?code=hello',
+        )
         self.assertEqual(token, {'access_token': 'urlencoded', 'token_type': 'Bearer'})
 
 
-class LinkedInComplianceFixTest(unittest.TestCase):
+class LinkedInComplianceFixTest(TestCase):
+
+    def setUp(self):
+        mocker = requests_mock.Mocker()
+        mocker.post(
+            "https://www.linkedin.com/uas/oauth2/accessToken",
+            json={"access_token": "linkedin"},
+        )
+        mocker.post(
+            "https://api.linkedin.com/v1/people/~/shares",
+            status_code=201,
+            json={
+              "updateKey": "UPDATE-3346389-595113200",
+              "updateUrl": "https://www.linkedin.com/updates?discuss=abc&scope=xyz"
+            }
+        )
+        mocker.start()
+        self.addCleanup(mocker.stop)
 
-    def test_fetch_access_token(self):
         linkedin = OAuth2Session('foo', redirect_uri='https://i.b')
-        linkedin = linkedin_compliance_fix(linkedin)
-
-        linkedin.post = mock.MagicMock()
-        response = requests.Response()
-        response.status_code = 200
-        response.request = mock.MagicMock()
-        response._content = '{"access_token":"linkedin"}'.encode('UTF-8')
-        linkedin.post.return_value = response
-
-        token = linkedin.fetch_token('https://mocked.out',
-                                     client_secret='bar',
-                                     authorization_response='https://i.b/?code=hello')
+        self.session = linkedin_compliance_fix(linkedin)
+
+    def test_fetch_access_token(self):
+        token = self.session.fetch_token(
+            'https://www.linkedin.com/uas/oauth2/accessToken',
+            client_secret='bar',
+            authorization_response='https://i.b/?code=hello',
+        )
         self.assertEqual(token, {'access_token': 'linkedin', 'token_type': 'Bearer'})
 
+    def test_protected_request(self):
+        self.session.token = {"access_token": 'dummy-access-token'}
+        response = self.session.post(
+            "https://api.linkedin.com/v1/people/~/shares"
+        )
+        url = response.request.url
+        query = parse_qs(urlparse(url).query)
+        self.assertEqual(query["oauth2_access_token"], ["dummy-access-token"])
 
-class MailChimpComplianceFixTest(unittest.TestCase):
 
-    def test_fetch_access_token(self):
+class MailChimpComplianceFixTest(TestCase):
+
+    def setUp(self):
+        mocker = requests_mock.Mocker()
+        mocker.post(
+            "https://login.mailchimp.com/oauth2/token",
+            json={"access_token": "mailchimp", "expires_in": 0, "scope": None},
+        )
+        mocker.start()
+        self.addCleanup(mocker.stop)
+
         mailchimp = OAuth2Session('foo', redirect_uri='https://i.b')
-        mailchimp = mailchimp_compliance_fix(mailchimp)
-
-        mailchimp.post = mock.MagicMock()
-        response = requests.Response()
-        response.status_code = 200
-        response.request = mock.MagicMock()
-        response._content = '{"access_token":"mailchimp", "expires_in":0, "scope":null}'.encode('UTF-8')
-        mailchimp.post.return_value = response
-
-        token = mailchimp.fetch_token('https://mocked.out',
-                                      client_secret='bar',
-                                      authorization_response='https://i.b/?code=hello')
+        self.session = mailchimp_compliance_fix(mailchimp)
+
+    def test_fetch_access_token(self):
+        token = self.session.fetch_token(
+            "https://login.mailchimp.com/oauth2/token",
+            client_secret='bar',
+            authorization_response='https://i.b/?code=hello',
+        )
         # Times should be close
         approx_expires_at = time.time() + 3600
         actual_expires_at = token.pop('expires_at')
@@ -76,24 +113,65 @@ def test_fetch_access_token(self):
         self.assertEqual(token, {'access_token': 'mailchimp', 'expires_in': 3600})
 
         # And no scope at all
-        self.assertFalse('scope' in token)
+        self.assertNotIn('scope', token)
 
 
-class WeiboComplianceFixTest(unittest.TestCase):
+class WeiboComplianceFixTest(TestCase):
+
+    def setUp(self):
+        mocker = requests_mock.Mocker()
+        mocker.post(
+            "https://api.weibo.com/oauth2/access_token",
+            json={"access_token": "weibo"},
+        )
+        mocker.start()
+        self.addCleanup(mocker.stop)
 
-    def test_fetch_access_token(self):
         weibo = OAuth2Session('foo', redirect_uri='https://i.b')
-        weibo = weibo_compliance_fix(weibo)
-
-        weibo.post = mock.MagicMock()
-        response = requests.Response()
-        response.status_code = 200
-        response.request = mock.MagicMock()
-        response._content = '{"access_token":"weibo"}'.encode('UTF-8')
-        weibo.post.return_value = response
-
-        token = weibo.fetch_token('https://mocked.out',
-                                     client_secret='bar',
-                                     authorization_response='https://i.b/?code=hello')
+        self.session = weibo_compliance_fix(weibo)
+
+    def test_fetch_access_token(self):
+        token = self.session.fetch_token(
+            'https://api.weibo.com/oauth2/access_token',
+            client_secret='bar',
+            authorization_response='https://i.b/?code=hello',
+        )
         self.assertEqual(token, {'access_token': 'weibo', 'token_type': 'Bearer'})
 
+
+class SlackComplianceFixTest(TestCase):
+
+    def setUp(self):
+        mocker = requests_mock.Mocker()
+        mocker.post(
+            "https://slack.com/api/oauth.access",
+            json={
+              "access_token": "xoxt-23984754863-2348975623103",
+              "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",
+            }
+        )
+        mocker.start()
+        self.addCleanup(mocker.stop)
+
+        slack = OAuth2Session('foo', redirect_uri='https://i.b')
+        self.session = slack_compliance_fix(slack)
+
+    def test_protected_request(self):
+        self.session.token = {"access_token": 'dummy-access-token'}
+        response = self.session.get(
+            "https://slack.com/api/auth.test"
+        )
+        url = response.request.url
+        query = parse_qs(urlparse(url).query)
+        self.assertEqual(query["token"], ["dummy-access-token"])