Async Flask for oauth during testing

async_oauth.py

import json
import os
from multiprocessing import Process, Queue

from flask import Flask, request
from requests_oauthlib import OAuth2Session
from pytest import fixture

app = Flask(__name__)


class AsyncOAuth:
    @staticmethod
    @app.route('/auto-close')
    def index():
        app.queue.put(request.url)
        return 'Close me!'

    @staticmethod
    def run_async():
        def run_flask(app_object):
            app_object.run(host='0.0.0.0', port=8080, ssl_context='adhoc')

        q = Queue()
        app.queue = q
        p = Process(target=run_flask, args=(app,))
        p.start()
        return q, p

      
@fixture(scope='function')
def oauth(pytestconfig):
    client_id = os.getenv('OAUTH_CLIENT_ID')
    client_secret = os.getenv('OAUTH_CLIENT_SECRET')
    scopes = os.getenv('OAUTH_SCOPES')
    token_path = Path.cwd() / 'token.json'
    if token_path.exists():
        with open(token_path, 'r') as token_file:
            token = json.loads(token_file.readline())
        
        oauth = OAuth2Session(
            client_id=client_id,
            token=token,
            auto_refresh_kwargs={
                'client_id': client_id,
                'client_secret': client_secret
            },
            auto_refresh_url='https://accounts.google.com/o/oauth2/token'
        )

        def updated_wrapper(oauth_session):
            def token_updater(updated_token):
                with open(token_path, 'w') as updated_token_file:
                    updated_token_file.write(json.dumps(updated_token))
                oauth_session.token = updated_token
            return token_updater
        oauth.token_updater = updated_wrapper(oauth)
    else:
        if pytestconfig:
            capmanager = pytestconfig.pluginmanager.getplugin('capturemanager')
        oauth = OAuth2Session(
            client_id=client_id,
            redirect_uri='https://127.0.0.1:8080/auto-close',
            scope=scopes
        )
        authorization_url, state = oauth.authorization_url(
            'https://accounts.google.com/o/oauth2/auth',
            access_type='offline',
            prompt='consent'
        )
        q, p = AsyncOAuth.run_async()
        if pytestconfig:
            capmanager.suspend_global_capture(in_=True)
        print('\n\nPlease go to {authorization_url} and authorize access.'.format(authorization_url=authorization_url)
        if pytestconfig:
            capmanager.resume_global_capture()
        oauth.fetch_token(
            'https://accounts.google.com/o/oauth2/token',
            authorization_response=q.get(),
            client_secret=client_secret
        )
        q.close()
        p.terminate()
        with open(token_path, 'w') as token_file:
            token_file.write(json.dumps(oauth.token))
    return oauth