<s> from __future__ import unicode_literals <EOL> import json <EOL> import mock <EOL> import time <EOL> from copy import deepcopy <EOL> try : <EOL> from unittest2 import TestCase <EOL> except ImportError : <EOL> from unittest import TestCase <EOL> from oauthlib . common import urlencode <EOL> from oauthlib . oauth2 import TokenExpiredError , OAuth2Error <EOL> from oauthlib . oauth2 import MismatchingStateError <EOL> from oauthlib . oauth2 import WebApplicationClient , MobileApplicationClient <EOL> from oauthlib . oauth2 import LegacyApplicationClient , BackendApplicationClient <EOL> from requests_oauthlib import OAuth2Session , TokenUpdated <EOL> fake_time = time . time ( ) <EOL> def fake_token ( token ) : <EOL> def fake_send ( r , ** kwargs ) : <EOL> resp = mock . MagicMock ( ) <EOL> resp . text = json . dumps ( token ) <EOL> return resp <EOL> return fake_send <EOL> class OAuth2SessionTest ( TestCase ) : <EOL> def setUp ( self ) : <EOL> if not hasattr ( self , '<STR_LIT>' ) : <EOL> self . assertIn = lambda a , b : self . assertTrue ( a in b ) <EOL> self . token = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : fake_time + <NUM_LIT> , <EOL> } <EOL> self . client_id = '<STR_LIT:foo>' <EOL> self . clients = [ <EOL> WebApplicationClient ( self . client_id , code = '<STR_LIT>' ) , <EOL> LegacyApplicationClient ( self . client_id ) , <EOL> BackendApplicationClient ( self . client_id ) , <EOL> ] <EOL> self . all_clients = self . clients + [ MobileApplicationClient ( self . client_id ) ] <EOL> def test_add_token ( self ) : <EOL> token = '<STR_LIT>' + self . token [ '<STR_LIT>' ] <EOL> def verifier ( r , ** kwargs ) : <EOL> auth_header = r . headers . get ( '<STR_LIT>' , None ) <EOL> if '<STR_LIT>' . encode ( '<STR_LIT:utf-8>' ) in r . headers : <EOL> auth_header = r . headers [ '<STR_LIT>' . encode ( '<STR_LIT:utf-8>' ) ] <EOL> self . assertEqual ( auth_header , token ) <EOL> resp = mock . MagicMock ( ) <EOL> resp . cookes = [ ] <EOL> return resp <EOL> for client in self . all_clients : <EOL> auth = OAuth2Session ( client = client , token = self . token ) <EOL> auth . send = verifier <EOL> auth . get ( '<STR_LIT>' ) <EOL> def test_authorization_url ( self ) : <EOL> url = '<STR_LIT>' <EOL> web = WebApplicationClient ( self . client_id ) <EOL> s = OAuth2Session ( client = web ) <EOL> auth_url , state = s . authorization_url ( url ) <EOL> self . assertIn ( state , auth_url ) <EOL> self . assertIn ( self . client_id , auth_url ) <EOL> self . assertIn ( '<STR_LIT>' , auth_url ) <EOL> mobile = MobileApplicationClient ( self . client_id ) <EOL> s = OAuth2Session ( client = mobile ) <EOL> auth_url , state = s . authorization_url ( url ) <EOL> self . assertIn ( state , auth_url ) <EOL> self . assertIn ( self . client_id , auth_url ) <EOL> self . assertIn ( '<STR_LIT>' , auth_url ) <EOL> @ mock . patch ( "<STR_LIT>" , new = lambda : fake_time ) <EOL> def test_refresh_token_request ( self ) : <EOL> self . expired_token = dict ( self . token ) <EOL> self . expired_token [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> del self . expired_token [ '<STR_LIT>' ] <EOL> def fake_refresh ( r , ** kwargs ) : <EOL> resp = mock . MagicMock ( ) <EOL> resp . text = json . dumps ( self . token ) <EOL> return resp <EOL> for client in self . clients : <EOL> auth = OAuth2Session ( client = client , token = self . expired_token ) <EOL> self . assertRaises ( TokenExpiredError , auth . get , '<STR_LIT>' ) <EOL> for client in self . clients : <EOL> auth = OAuth2Session ( client = client , token = self . expired_token , <EOL> auto_refresh_url = '<STR_LIT>' ) <EOL> auth . send = fake_refresh <EOL> self . assertRaises ( TokenUpdated , auth . get , '<STR_LIT>' ) <EOL> def token_updater ( token ) : <EOL> self . assertEqual ( token , self . token ) <EOL> for client in self . clients : <EOL> auth = OAuth2Session ( client = client , token = self . expired_token , <EOL> auto_refresh_url = '<STR_LIT>' , <EOL> token_updater = token_updater ) <EOL> auth . send = fake_refresh <EOL> auth . get ( '<STR_LIT>' ) <EOL> @ mock . patch ( "<STR_LIT>" , new = lambda : fake_time ) <EOL> def test_token_from_fragment ( self ) : <EOL> mobile = MobileApplicationClient ( self . client_id ) <EOL> response_url = '<STR_LIT>' + urlencode ( self . token . items ( ) ) <EOL> auth = OAuth2Session ( client = mobile ) <EOL> self . assertEqual ( auth . token_from_fragment ( response_url ) , self . token ) <EOL> @ mock . patch ( "<STR_LIT>" , new = lambda : fake_time ) <EOL> def test_fetch_token ( self ) : <EOL> url = '<STR_LIT>' <EOL> for client in self . clients : <EOL> auth = OAuth2Session ( client = client , token = self . token ) <EOL> auth . send = fake_token ( self . token ) <EOL> self . assertEqual ( auth . fetch_token ( url ) , self . token ) <EOL> error = { '<STR_LIT:error>' : '<STR_LIT>' } <EOL> for client in self . clients : <EOL> auth = OAuth2Session ( client = client , token = self . token ) <EOL> auth . send = fake_token ( error ) <EOL> self . assertRaises ( OAuth2Error , auth . fetch_token , url ) <EOL> def test_cleans_previous_token_before_fetching_new_one ( self ) : <EOL> """<STR_LIT>""" <EOL> new_token = deepcopy ( self . token ) <EOL> past = time . time ( ) - <NUM_LIT> <EOL> now = time . time ( ) <EOL> self . token [ '<STR_LIT>' ] = past <EOL> new_token [ '<STR_LIT>' ] = now + <NUM_LIT> <EOL> url = '<STR_LIT>' <EOL> with mock . patch ( '<STR_LIT>' , lambda : now ) : <EOL> for client in self . clients : <EOL> auth = OAuth2Session ( client = client , token = self . token ) <EOL> auth . send = fake_token ( new_token ) <EOL> self . assertEqual ( auth . fetch_token ( url ) , new_token ) <EOL> def test_web_app_fetch_token ( self ) : <EOL> client = OAuth2Session ( '<STR_LIT:foo>' , state = '<STR_LIT>' ) <EOL> self . assertRaises ( MismatchingStateError , client . fetch_token , <EOL> '<STR_LIT>' , <EOL> authorization_response = '<STR_LIT>' ) <EOL> def test_client_id_proxy ( self ) : <EOL> sess = OAuth2Session ( '<STR_LIT>' ) <EOL> self . assertEqual ( sess . client_id , '<STR_LIT>' ) <EOL> sess . client_id = '<STR_LIT>' <EOL> self . assertEqual ( sess . client_id , '<STR_LIT>' ) <EOL> sess . _client . client_id = '<STR_LIT>' <EOL> self . assertEqual ( sess . client_id , '<STR_LIT>' ) <EOL> del sess . client_id <EOL> self . assertIsNone ( sess . client_id ) <EOL> def test_access_token_proxy ( self ) : <EOL> sess = OAuth2Session ( '<STR_LIT>' ) <EOL> self . assertIsNone ( sess . access_token ) <EOL> sess . access_token = '<STR_LIT>' <EOL> self . assertEqual ( sess . access_token , '<STR_LIT>' ) <EOL> sess . _client . access_token = '<STR_LIT>' <EOL> self . assertEqual ( sess . access_token , '<STR_LIT>' ) <EOL> del sess . access_token <EOL> self . assertIsNone ( sess . access_token ) <EOL> def test_token_proxy ( self ) : <EOL> token = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> sess = OAuth2Session ( '<STR_LIT>' , token = token ) <EOL> self . assertEqual ( sess . access_token , '<STR_LIT>' ) <EOL> self . assertEqual ( sess . token , token ) <EOL> token [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> sess . token = token <EOL> self . assertEqual ( sess . access_token , '<STR_LIT>' ) <EOL> self . assertEqual ( sess . token , token ) <EOL> sess . _client . access_token = '<STR_LIT>' <EOL> token [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> self . assertEqual ( sess . access_token , '<STR_LIT>' ) <EOL> self . assertEqual ( sess . token , token ) <EOL> with self . assertRaises ( AttributeError ) : <EOL> del sess . token <EOL> def test_authorized_false ( self ) : <EOL> sess = OAuth2Session ( '<STR_LIT:foo>' ) <EOL> self . assertFalse ( sess . authorized ) <EOL> @ mock . patch ( "<STR_LIT>" , new = lambda : fake_time ) <EOL> def test_authorized_true ( self ) : <EOL> def fake_token ( token ) : <EOL> def fake_send ( r , ** kwargs ) : <EOL> resp = mock . MagicMock ( ) <EOL> resp . text = json . dumps ( token ) <EOL> return resp <EOL> return fake_send <EOL> url = '<STR_LIT>' <EOL> for client in self . clients : <EOL> sess = OAuth2Session ( client = client ) <EOL> sess . send = fake_token ( self . token ) <EOL> self . assertFalse ( sess . authorized ) <EOL> sess . fetch_token ( url ) <EOL> self . assertTrue ( sess . authorized ) </s>
