add a fitbit compliance fix

requests_oauthlib/compliance_fixes/__init__.py

@@ -1,6 +1,7 @@
 from __future__ import absolute_import
 
 from .facebook import facebook_compliance_fix
+from .fitbit import fitbit_compliance_fix
 from .linkedin import linkedin_compliance_fix
 from .slack import slack_compliance_fix
 from .mailchimp import mailchimp_compliance_fix

requests_oauthlib/compliance_fixes/fitbit.py

@@ -0,0 +1,26 @@
+"""
+The Fitbit API breaks from the OAuth2 RFC standard by returning an "errors"
+object list, rather than a single "error" string. This puts hooks in place so
+that oauthlib can process an error in the results from access token and refresh
+token responses. This is necessary to prevent getting the generic red herring
+MissingTokenError.
+"""
+
+from json import loads, dumps
+
+from oauthlib.common import to_unicode
+
+
+def fitbit_compliance_fix(session):
+
+    def _missing_error(r):
+        token = loads(r.text)
+        if 'errors' in token:
+            # Set the error to the first one we have
+            token['error'] = token['errors'][0]['errorType']
+        r._content = to_unicode(dumps(token)).encode('UTF-8')
+        return r
+
+    session.register_compliance_hook('access_token_response', _missing_error)
+    session.register_compliance_hook('refresh_token_response', _missing_error)
+    return session

tests/test_compliance_fixes.py

@@ -12,8 +12,10 @@
 except ImportError:
     from urllib.parse import urlparse, parse_qs
 
+from oauthlib.oauth2.rfc6749.errors import InvalidGrantError
 from requests_oauthlib import OAuth2Session
 from requests_oauthlib.compliance_fixes import facebook_compliance_fix
+from requests_oauthlib.compliance_fixes import fitbit_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
@@ -44,6 +46,63 @@ def test_fetch_access_token(self):
         self.assertEqual(token, {'access_token': 'urlencoded', 'token_type': 'Bearer'})
 
 
+class FitbitComplianceFixTest(TestCase):
+
+    def setUp(self):
+        self.mocker = requests_mock.Mocker()
+        self.mocker.post(
+            "https://api.fitbit.com/oauth2/token",
+            json={"errors": [{"errorType": "invalid_grant"}]},
+        )
+        self.mocker.start()
+        self.addCleanup(self.mocker.stop)
+
+        fitbit = OAuth2Session('foo', redirect_uri='https://i.b')
+        self.session = fitbit_compliance_fix(fitbit)
+
+    def test_fetch_access_token(self):
+        self.assertRaises(
+            InvalidGrantError,
+            self.session.fetch_token,
+            'https://api.fitbit.com/oauth2/token',
+            client_secret='bar',
+            authorization_response='https://i.b/?code=hello',
+        )
+
+        self.mocker.post(
+            "https://api.fitbit.com/oauth2/token",
+            json={"access_token": "fitbit"},
+        )
+
+        token = self.session.fetch_token(
+            'https://api.fitbit.com/oauth2/token',
+            client_secret='good'
+        )
+
+        self.assertEqual(token, {'access_token': 'fitbit'})
+
+    def test_refresh_token(self):
+        self.assertRaises(
+            InvalidGrantError,
+            self.session.refresh_token,
+            'https://api.fitbit.com/oauth2/token',
+            auth=requests.auth.HTTPBasicAuth('foo', 'bar')
+        )
+
+        self.mocker.post(
+            "https://api.fitbit.com/oauth2/token",
+            json={"access_token": "access", "refresh_token": "refresh"},
+        )
+
+        token = self.session.refresh_token(
+            'https://api.fitbit.com/oauth2/token',
+            auth=requests.auth.HTTPBasicAuth('foo', 'bar')
+        )
+
+        self.assertEqual(token['access_token'], 'access')
+        self.assertEqual(token['refresh_token'], 'refresh')
+
+
 class LinkedInComplianceFixTest(TestCase):
 
     def setUp(self):