# Path: oauthenticator/azuread.py
# PYJWT_2 = V(jwt.__version__) >= V("2.0")
# 
# Path: oauthenticator/mediawiki.py
# AUTH_REQUEST_COOKIE_NAME = 'mw_oauth_request_token_v2'
# 
# Path: oauthenticator/mediawiki.py
# class MWOAuthenticator(OAuthenticator):
#     login_service = 'MediaWiki'
#     login_handler = MWLoginHandler
#     callback_handler = MWCallbackHandler
# 
#     mw_index_url = Unicode(
#         os.environ.get('MW_INDEX_URL', 'https://meta.wikimedia.org/w/index.php'),
#         config=True,
#         help='Full path to index.php of the MW instance to use to log in',
#     )
# 
#     executor_threads = Integer(
#         12,
#         help="""Number of executor threads.
# 
#         MediaWiki OAuth requests happen in this thread,
#         so it is mostly waiting for network replies.
#         """,
#         config=True,
#     )
#     executor = Any()
# 
#     def normalize_username(self, username):
#         """
#         Override normalize_username to avoid lowercasing usernames
#         """
#         return username
# 
#     def _executor_default(self):
#         return ThreadPoolExecutor(self.executor_threads)
# 
#     async def authenticate(self, handler, data=None):
#         consumer_token = ConsumerToken(
#             self.client_id,
#             self.client_secret,
#         )
# 
#         handshaker = Handshaker(self.mw_index_url, consumer_token)
#         request_token = dejsonify(handler.get_secure_cookie(AUTH_REQUEST_COOKIE_NAME))
#         handler.clear_cookie(AUTH_REQUEST_COOKIE_NAME)
#         access_token = await wrap_future(
#             self.executor.submit(
#                 handshaker.complete, request_token, handler.request.query
#             )
#         )
# 
#         identity = await wrap_future(
#             self.executor.submit(handshaker.identify, access_token)
#         )
#         if identity and 'username' in identity:
#             # this shouldn't be necessary anymore,
#             # but keep for backward-compatibility
#             return {
#                 'name': identity['username'].replace(' ', '_'),
#                 'auth_state': {
#                     'ACCESS_TOKEN_KEY': access_token.key,
#                     'ACCESS_TOKEN_SECRET': access_token.secret,
#                     'MEDIAWIKI_USER_IDENTITY': identity,
#                 },
#             }
#         else:
#             self.log.error("No username found in %s", identity)
# 
# Path: oauthenticator/tests/mocks.py
# def mock_handler(Handler, uri='https://hub.example.com', method='GET', **settings):
#     """Instantiate a Handler in a mock application"""
#     application = web.Application(
#         hub=Mock(
#             base_url='/hub/',
#             server=Mock(base_url='/hub/'),
#         ),
#         cookie_secret=os.urandom(32),
#         db=Mock(rollback=Mock(return_value=None)),
#         **settings
#     )
#     request = HTTPServerRequest(
#         method=method,
#         uri=uri,
#         connection=Mock(),
#     )
#     handler = Handler(
#         application=application,
#         request=request,
#     )
#     handler._transforms = []
#     return handler


import json
import re
import time
import jwt
import requests_mock
from unittest.mock import Mock
from pytest import fixture
from tornado import web
from ..azuread import PYJWT_2
from ..mediawiki import AUTH_REQUEST_COOKIE_NAME
from ..mediawiki import MWOAuthenticator
from .mocks import mock_handler
        mock.post('/w/index.php?title=Special%3AOAuth%2Fidentify', content=post_token)
        yield mock


def new_authenticator():
    return MWOAuthenticator(
        client_id='client_id',
        client_secret='client_secret',
    )


async def test_mediawiki(mediawiki):
    authenticator = new_authenticator()
    handler = Mock(
        spec=web.RequestHandler,
        get_secure_cookie=Mock(return_value=json.dumps(['key', 'secret'])),
        request=Mock(query='oauth_token=key&oauth_verifier=me'),
    )
    user = await authenticator.authenticate(handler, None)
    assert user['name'] == 'wash'
    auth_state = user['auth_state']
    assert auth_state['ACCESS_TOKEN_KEY'] == 'key'
    assert auth_state['ACCESS_TOKEN_SECRET'] == 'secret'
    identity = auth_state['MEDIAWIKI_USER_IDENTITY']
    assert identity['username'] == user['name']


async def test_login_redirect(mediawiki):
    authenticator = new_authenticator()
    record = []

# Path: oauthenticator/azuread.py
# PYJWT_2 = V(jwt.__version__) >= V("2.0")
# 
# Path: oauthenticator/mediawiki.py
# AUTH_REQUEST_COOKIE_NAME = 'mw_oauth_request_token_v2'
# 
# Path: oauthenticator/mediawiki.py
# class MWOAuthenticator(OAuthenticator):
#     login_service = 'MediaWiki'
#     login_handler = MWLoginHandler
#     callback_handler = MWCallbackHandler
# 
#     mw_index_url = Unicode(
#         os.environ.get('MW_INDEX_URL', 'https://meta.wikimedia.org/w/index.php'),
#         config=True,
#         help='Full path to index.php of the MW instance to use to log in',
#     )
# 
#     executor_threads = Integer(
#         12,
#         help="""Number of executor threads.
# 
#         MediaWiki OAuth requests happen in this thread,
#         so it is mostly waiting for network replies.
#         """,
#         config=True,
#     )
#     executor = Any()
# 
#     def normalize_username(self, username):
#         """
#         Override normalize_username to avoid lowercasing usernames
#         """
#         return username
# 
#     def _executor_default(self):
#         return ThreadPoolExecutor(self.executor_threads)
# 
#     async def authenticate(self, handler, data=None):
#         consumer_token = ConsumerToken(
#             self.client_id,
#             self.client_secret,
#         )
# 
#         handshaker = Handshaker(self.mw_index_url, consumer_token)
#         request_token = dejsonify(handler.get_secure_cookie(AUTH_REQUEST_COOKIE_NAME))
#         handler.clear_cookie(AUTH_REQUEST_COOKIE_NAME)
#         access_token = await wrap_future(
#             self.executor.submit(
#                 handshaker.complete, request_token, handler.request.query
#             )
#         )
# 
#         identity = await wrap_future(
#             self.executor.submit(handshaker.identify, access_token)
#         )
#         if identity and 'username' in identity:
#             # this shouldn't be necessary anymore,
#             # but keep for backward-compatibility
#             return {
#                 'name': identity['username'].replace(' ', '_'),
#                 'auth_state': {
#                     'ACCESS_TOKEN_KEY': access_token.key,
#                     'ACCESS_TOKEN_SECRET': access_token.secret,
#                     'MEDIAWIKI_USER_IDENTITY': identity,
#                 },
#             }
#         else:
#             self.log.error("No username found in %s", identity)
# 
# Path: oauthenticator/tests/mocks.py
# def mock_handler(Handler, uri='https://hub.example.com', method='GET', **settings):
#     """Instantiate a Handler in a mock application"""
#     application = web.Application(
#         hub=Mock(
#             base_url='/hub/',
#             server=Mock(base_url='/hub/'),
#         ),
#         cookie_secret=os.urandom(32),
#         db=Mock(rollback=Mock(return_value=None)),
#         **settings
#     )
#     request = HTTPServerRequest(
#         method=method,
#         uri=uri,
#         connection=Mock(),
#     )
#     handler = Handler(
#         application=application,
#         request=request,
#     )
#     handler._transforms = []
#     return handler


# Path: oauthenticator/tests/test_mediawiki.py
import json
import re
import time
import jwt
import requests_mock
from unittest.mock import Mock
from pytest import fixture
from tornado import web
from ..azuread import PYJWT_2
from ..mediawiki import AUTH_REQUEST_COOKIE_NAME
from ..mediawiki import MWOAuthenticator
from .mocks import mock_handler

        mock.post('/w/index.php?title=Special%3AOAuth%2Fidentify', content=post_token)
        yield mock


def new_authenticator():
    return MWOAuthenticator(
        client_id='client_id',
        client_secret='client_secret',
    )


async def test_mediawiki(mediawiki):
    authenticator = new_authenticator()
    handler = Mock(
        spec=web.RequestHandler,
        get_secure_cookie=Mock(return_value=json.dumps(['key', 'secret'])),
        request=Mock(query='oauth_token=key&oauth_verifier=me'),
    )
    user = await authenticator.authenticate(handler, None)
    assert user['name'] == 'wash'
    auth_state = user['auth_state']
    assert auth_state['ACCESS_TOKEN_KEY'] == 'key'
    assert auth_state['ACCESS_TOKEN_SECRET'] == 'secret'
    identity = auth_state['MEDIAWIKI_USER_IDENTITY']
    assert identity['username'] == user['name']


async def test_login_redirect(mediawiki):
    authenticator = new_authenticator()
    record = []

    handler = mock_handler(