Merge pull request #362 from AdamGold/feature/instagram-compliance-fix

add instagram compliance fix

Author: singingwolfboy

requests_oauthlib/compliance_fixes/__init__.py

@@ -4,6 +4,7 @@
 from .fitbit import fitbit_compliance_fix
 from .linkedin import linkedin_compliance_fix
 from .slack import slack_compliance_fix
+from .instagram import instagram_compliance_fix
 from .mailchimp import mailchimp_compliance_fix
 from .weibo import weibo_compliance_fix
 from .plentymarkets import plentymarkets_compliance_fix

requests_oauthlib/compliance_fixes/instagram.py

@@ -0,0 +1,27 @@
+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 instagram_compliance_fix(session):
+    def _non_compliant_param_name(url, headers, data):
+        # If the user has already specified the token in the URL
+        # 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("access_token")
+        if token:
+            # Nothing to do, just return.
+            return url, headers, data
+
+        token = [('access_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)
+    return session

tests/test_compliance_fixes.py

@@ -17,6 +17,7 @@
 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
+from requests_oauthlib.compliance_fixes import instagram_compliance_fix
 from requests_oauthlib.compliance_fixes import plentymarkets_compliance_fix
 
 
@@ -275,6 +276,58 @@ def test_protected_request_override_token_url(self):
         self.assertIsNone(response.request.body)
 
 
+class InstagramComplianceFixTest(TestCase):
+
+    def setUp(self):
+        mocker = requests_mock.Mocker()
+        mocker.request(
+            method="GET",
+            url="https://api.instagram.com/v1/users/self",
+            json={
+                "data": {
+                    "id": "1574083",
+                    "username": "snoopdogg",
+                    "full_name": "Snoop Dogg",
+                    "profile_picture": "http://distillery.s3.amazonaws.com/profiles/profile_1574083_75sq_1295469061.jpg",
+                    "bio": "This is my bio",
+                    "website": "http://snoopdogg.com",
+                    "is_business": False,
+                    "counts": {
+                        "media": 1320,
+                        "follows": 420,
+                        "followed_by": 3410
+                    }
+                }
+            }
+        )
+        mocker.start()
+        self.addCleanup(mocker.stop)
+
+        instagram = OAuth2Session('someclientid', redirect_uri='https://i.b')
+        self.session = instagram_compliance_fix(instagram)
+
+    def test_protected_request(self):
+        self.session.token = {"access_token": 'dummy-access-token'}
+        response = self.session.get(
+            "https://api.instagram.com/v1/users/self"
+        )
+        url = response.request.url
+        query = parse_qs(urlparse(url).query)
+        self.assertIn("access_token", query)
+        self.assertEqual(query["access_token"], ["dummy-access-token"])
+
+    def test_protected_request_dont_override(self):
+        """check that if the access_token param
+        already exist we don't override it"""
+        self.session.token = {"access_token": 'dummy-access-token'}
+        response = self.session.get(
+            "https://api.instagram.com/v1/users/self?access_token=correct-access-token"
+        )
+        url = response.request.url
+        query = parse_qs(urlparse(url).query)
+        self.assertIn("access_token", query)
+        self.assertEqual(query["access_token"], ["correct-access-token"])
+
 class PlentymarketsComplianceFixTest(TestCase):
 
     def setUp(self):