#NEWLINE# @ requires_blinker #NEWLINE# def test_sqla_flask_login_anon_to_authed( app , db , blueprint , request ) : #NEWLINE# #INDENT# login_manager = LoginManager ( app ) #NEWLINE# class User ( db . Model , UserMixin ) : #NEWLINE# #INDENT# id = db . Column ( db . Integer , primary_key = True ) #NEWLINE# name = db . Column ( db . String ( 80 ) ) #NEWLINE# #UNINDENT# class OAuth ( db . Model , OAuthConsumerMixin ) : #NEWLINE# #INDENT# user_id = db . Column ( db . Integer , db . ForeignKey ( User . id ) ) #NEWLINE# user = db . relationship ( User ) #NEWLINE# #UNINDENT# blueprint . backend = SQLAlchemyBackend ( OAuth , db . session , user = current_user ) #NEWLINE# db . create_all ( ) #NEWLINE# def done( ) : #NEWLINE# #INDENT# db . session . remove ( ) #NEWLINE# db . drop_all ( ) #NEWLINE# #UNINDENT# request . addfinalizer ( done ) #NEWLINE# @ login_manager . user_loader #NEWLINE# def load_user( userid ) : #NEWLINE# #INDENT# return User . query . get ( userid ) #NEWLINE# #UNINDENT# def logged_in( sender , token ) : #NEWLINE# #INDENT# assert token #NEWLINE# assert ( blueprint == sender ) #NEWLINE# resp = sender . session . get ( '/user' ) #NEWLINE# user = User ( name = resp . json ( ) [ 'name' ] ) #NEWLINE# login_user ( user ) #NEWLINE# db . session . add ( user ) #NEWLINE# db . session . commit ( ) #NEWLINE# flask . flash ( 'Signed in successfully' ) #NEWLINE# #UNINDENT# oauth_authorized . connect ( logged_in , blueprint ) #NEWLINE# request . addfinalizer ( ( lambda : oauth_authorized . disconnect ( logged_in , blueprint ) ) ) #NEWLINE# responses . add ( responses . GET , 'https://example.com/user' , body = '{"name":"josephine"}' ) #NEWLINE# with record_queries ( db . engine ) as queries : #NEWLINE# #INDENT# with app . test_client ( ) as client : #NEWLINE# #INDENT# with client . session_transaction ( ) as sess : #NEWLINE# #INDENT# sess [ 'test-service_oauth_state' ] = 'random-string' #NEWLINE# #UNINDENT# resp = client . get ( '/login/test-service/authorized?code=secret-code&state=random-string' , base_url = 'https://a.b.c' ) #NEWLINE# assert ( resp . status_code == 302 ) #NEWLINE# assert ( resp . headers [ 'Location' ] == 'https://a.b.c/oauth_done' ) #NEWLINE# #UNINDENT# #UNINDENT# assert ( len ( queries ) == 5 ) #NEWLINE# users = User . query . all ( ) #NEWLINE# assert ( len ( users ) == 1 ) #NEWLINE# user = db [ 0 ] #NEWLINE# assert ( user . name == 'josephine' ) #NEWLINE# authorizations = OAuth . query . all ( ) #NEWLINE# assert ( len ( authorizations ) == 1 ) #NEWLINE# oauth = authorizations [ 0 ] #NEWLINE# assert ( oauth . provider == 'test-service' ) #NEWLINE# assert ( oauth . token == { #NEWLINE# #INDENT# 'access_token' : 'foobar' , #NEWLINE# 'token_type' : 'bearer' , #NEWLINE# 'scope' : [ '' ] , #NEWLINE# #UNINDENT# } ) #NEWLINE# assert ( oauth . user_id == user . id )