""" Tests for all OAuth2 errors. """
import json
import warnings

try:
    from __builtin__ import unichr as _unichr
except ImportError:
    _unichr = chr

try:
    from urlparse import urlparse
    from urllib import urlencode
except ImportError:
    # noinspection PyUnresolvedReferences
    from urllib.parse import urlparse, urlencode

from twisted.web.server import NOT_DONE_YET
from twisted.web.http import ACCEPTED, UNAUTHORIZED, FOUND
from txoauth2.errors import OAuth2Error, AuthorizationError

from tests import TwistedTestCase, MockRequest


class OAuth2ErrorTest(TwistedTestCase):
    """ Test for OAuth2Errors. """
    _VALID_CODE = ACCEPTED
    _VALID_NAME = 'Error name'
    _VALID_DESCRIPTION = 'Error description'
    _VALID_ERROR_URI = 'error:uri'
    _VALID_SCOPE = ['scope']

    @staticmethod
    def _initializeError(code, name, **kwargs):
        """
        Initialize a new error instance that should be tested.

        :param code: The return code of the error.
        :param name: The name of the error.
        :param kwargs: Additional arguments for the error.
        :return: The new initialized error.
        """
        return OAuth2Error(code, name, **kwargs)

    def assertValidErrorResponse(
            self, request, resultBytes, code=None, name=None, description=None, errorUri=None,
            scope=None, addWwwAuthenticateHeader=False, authScheme='Bearer'):
        """
        Assert that the response generated by the error matches the expected error response.

        :param request: The request that was modified by the error.
        :param resultBytes: The result that was returned by the error.
        :param code: The expected HTTP return code.
        :param name: The expected error name.
        :param description: The expected error description.
        :param errorUri: The expected error uri.
        :param scope: The expected scope.
        :param addWwwAuthenticateHeader: Whether or not the error should have added a
                                         WWW-Authenticate response header.
        :param authScheme: The expected authScheme in the response WWW-Authenticate header.
        """
        parameter = {'error': name, 'error_description': description,
                     'error_uri': errorUri, 'scope': scope}
        self.assertEqual(
            code, request.responseCode, msg='Expected the error to set the response code.')
        self.assertEqual(
            'application/json;charset=UTF-8', request.getResponseHeader('Content-Type'),
            msg='Expected the error to indicate that the response is utf-8 encoded json.')
        self.assertEqual(
            'no-store', request.getResponseHeader('Cache-Control'),
            msg='Expected the error to indicate that the response should not be cached.')
        self.assertEqual(
            'no-cache', request.getResponseHeader('Pragma'),
            msg='Expected the error to indicate that the response should not be cached.')
        if addWwwAuthenticateHeader:
            self.assertIsNotNone(
                request.getResponseHeader('WWW-Authenticate'),
                msg='Expected the error to include a WWW-Authenticate header.')
            requestAuthScheme, params = request.getResponseHeader('WWW-Authenticate').split(' ', 1)
            self.assertEqual(authScheme, requestAuthScheme,
                             msg='Expected the error to use the configured auth scheme '
                                 'in the WWW-Authenticate header.')
            authenticateParameter = dict(item.split('=', 1) for item in params.split(','))
            for parameterName, value in parameter.items():
                if value is None:
                    self.assertNotIn(parameterName, authenticateParameter,
                                     msg='Expected the error not to add the {name} parameter to '
                                         'the WWW-Authenticate header.'.format(name=parameterName))
                    continue
                self.assertIn(parameterName, authenticateParameter,
                              msg='Expected the error to add the {name} parameter '
                                  'to the WWW-Authenticate header.'.format(name=parameterName))
                self.assertEqual(
                    '"' + value + '"', authenticateParameter.pop(parameterName),
                    msg='Expected the error to include the correct value for the parameter {name} '
                        'in the WWW-Authenticate header.'.format(name=parameterName))
            self.assertIn('realm', authenticateParameter,
                          msg='Expected the error to add the realm to the WWW-Authenticate header.')
            self.assertEqual(
                '"{realm}"'.format(realm=request.prePathURL().decode('utf-8')),
                authenticateParameter.pop('realm'),
                msg='Expected the error to include the correct realm '
                    'in the WWW-Authenticate header.')
            self.assertDictEqual(
                authenticateParameter, {}, msg='Expected the error not to return any additional '
                                               'parameters in the WWW-Authenticate header.')
        else:
            self.assertIsNone(
                request.getResponseHeader('WWW-Authenticate'),
                msg='Expected the error not to include a WWW-Authenticate header.')
        self.assertIsInstance(
            resultBytes, bytes, message='Expected the error to generate bytes.')
        result = json.loads(resultBytes.decode('utf-8'))
        self.assertIsInstance(result, dict, message='Expected the error return a json object.')
        parameter['scope'] = None
        for parameterName, value in parameter.items():
            if value is None:
                self.assertNotIn(parameterName, result,
                                 msg='Expected the error not to add the {name} '
                                     'parameter to the response.'.format(name=parameterName))
                continue
            self.assertIn(parameterName, result,
                          msg='Expected the error to add the {name} parameter '
                              'to the response.'.format(name=parameterName))
            self.assertEqual(value, result.pop(parameterName),
                             msg='Expected the error to return the correct value '
                                 'for the parameter {name}.'.format(name=parameterName))
        self.assertDictEqual(
            result, {}, msg='Expected the error not to return any additional parameters.')

    def testInitialization(self):
        """ Test that the error can be initialized correctly and checks the argument types. """
        authScheme = 'Test'
        error = self._initializeError(
            self._VALID_CODE, self._VALID_NAME, description=self._VALID_DESCRIPTION,
            errorUri=self._VALID_ERROR_URI, scope=self._VALID_SCOPE,
            addWwwAuthenticateHeader=True, authScheme=authScheme)
        self.assertEqual(self._VALID_CODE, error.code,
                         msg='Expected the {error} to use the code that it has been '
                             'initialized with.'.format(error=error.__class__.__name__))
        self.assertEqual(self._VALID_NAME, error.name,
                         msg='Expected the {error} to use the name that it has been '
                             'initialized with.'.format(error=error.__class__.__name__))
        self.assertEqual(self._VALID_DESCRIPTION, error.description,
                         msg='Expected the {error} to use the description that it has been '
                             'initialized with.'.format(error=error.__class__.__name__))
        self.assertEqual(self._VALID_ERROR_URI, error.errorUri,
                         msg='Expected the {error} to use the error uri that it has been '
                             'initialized with.'.format(error=error.__class__.__name__))
        self.assertListEqual(error.scope, self._VALID_SCOPE,
                             msg='Expected the {error} to use the scope that it has been '
                                 'initialized with.'.format(error=error.__class__.__name__))
        self.assertTrue(error._addWwwAuthenticateHeader,  # pylint: disable=protected-access
                        msg='Expected the {error} to enable the WWW-Authenticate header.'.format(
                            error=error.__class__.__name__))
        self.assertEqual(authScheme, error._authScheme,  # pylint: disable=protected-access
                         msg='Expected the {error} to use the auth scheme that it has been '
                             'initialized with.'.format(error=error.__class__.__name__))
        error = self._initializeError(UNAUTHORIZED, self._VALID_NAME, scope=self._VALID_SCOPE[0])
        self.assertTrue(error._addWwwAuthenticateHeader,  # pylint: disable=protected-access
                        msg='Expected the {error} to enable the WWW-Authenticate header if the '
                            'code is 401 (UNAUTHORIZED).'.format(error=error.__class__.__name__))
        self.assertListEqual(error.scope, self._VALID_SCOPE,
                             msg='Expected the {error} to accept a single scope item.'
                             .format(error=error.__class__.__name__))
        error = self._initializeError(
            self._VALID_CODE, self._VALID_NAME, scope=[], addWwwAuthenticateHeader=False)
        self.assertFalse(error._addWwwAuthenticateHeader,  # pylint: disable=protected-access
                         msg='Expected the {error} not to enable the WWW-Authenticate header if it '
                             'is explicitly disabled.'.format(error=error.__class__.__name__))
        self.assertIsNone(error.scope, msg='Expected the {error} to treat an empty scope list as '
                                           'no scope.'.format(error=error.__class__.__name__))

    def testValidatesParameters(self):
        """ Test that invalid parameter values raise the appropriate exception. """
        for invalidCode in [None, '1', b'1', u'1', [], {}, object(), 2.0]:
            self.assertRaises(TypeError, self._initializeError, invalidCode, self._VALID_NAME)
        invalidValues = [1, [], {}, object(), 2.0]
        if not isinstance(b'', str):
            invalidValues.append(b'byteString')
        for invalidValue in invalidValues:
            self.assertRaises(TypeError, self._initializeError, self._VALID_CODE, invalidValue)
            self.assertRaises(TypeError, self._initializeError, self._VALID_CODE,
                              self._VALID_NAME, invalidValue)
            self.assertRaises(TypeError, self._initializeError, self._VALID_CODE,
                              self._VALID_NAME, errorUri=invalidValue)
            if not isinstance(invalidValue, list):
                self.assertRaises(TypeError, self._initializeError, self._VALID_CODE,
                                  self._VALID_NAME, scope=invalidValue)
            self.assertRaises(TypeError, self._initializeError, self._VALID_CODE,
                              self._VALID_NAME, scope=['valid', invalidValue])
            self.assertRaises(TypeError, self._initializeError, self._VALID_CODE,
                              self._VALID_NAME, authScheme=invalidValue)
        self.assertRaises(TypeError, self._initializeError, self._VALID_CODE, None,
                          self._VALID_DESCRIPTION, self._VALID_ERROR_URI)
        self.assertRaises(TypeError, self._initializeError, self._VALID_CODE,
                          self._VALID_NAME, addWwwAuthenticateHeader=object())
        invalidCharacters = [chr(char) for char in range(0x20)] + [
            _unichr(char) for char in range(0x7F, 0x120)] + ['"', '\\', '\xFF']
        for invalidCharacter in invalidCharacters:
            with warnings.catch_warnings(record=True) as caughtWarnings:
                warnings.simplefilter('always')
                error = self._initializeError(
                    self._VALID_CODE, self._VALID_NAME + invalidCharacter,
                    description=self._VALID_DESCRIPTION + invalidCharacter,
                    errorUri=self._VALID_ERROR_URI + invalidCharacter,
                    scope=['a' + invalidCharacter, 'b' + invalidCharacter])
                self.assertEqual(
                    4, len(caughtWarnings),
                    msg='Expected the {error} to generate 4 warnings for 4 invalid characters, if '
                        'it is initialized with a name, an error uri and two scopes containing an '
                        'invalid character.'.format(error=error.__class__.__name__))
                for warning in caughtWarnings:
                    self.assertTrue(issubclass(warning.category, RuntimeWarning),
                                    msg='Expected the {error} to generate a RuntimeWarning.'
                                    .format(error=error.__class__.__name__))
                    self.assertIn(
                        'Invalid character {char!r} in parameter'.format(char=invalidCharacter),
                        str(warning.message),
                        msg='Expected the warning to explain, that a character is invalid: '
                            '{warning}'.format(warning=warning.message))
            self.assertEqual(self._VALID_NAME + '?', error.name,
                             msg='Expected the error to escape the name parameter.')
            self.assertEqual(self._VALID_DESCRIPTION + '?', error.description,
                             msg='Expected the error to escape the description parameter.')
            self.assertEqual(self._VALID_ERROR_URI + '?', error.errorUri,
                             msg='Expected the error to escape the errorUri parameter.')
            self.assertListEqual(['a?', 'b?'], error.scope,
                                 msg='Expected the error to escape the scope parameter.')
        with warnings.catch_warnings(record=True) as caughtWarnings:
            warnings.simplefilter('always')
            error = self._initializeError(
                self._VALID_CODE, self._VALID_NAME,
                description=self._VALID_DESCRIPTION, errorUri=self._VALID_ERROR_URI + ' ')
            self.assertEqual(
                1, len(caughtWarnings),
                msg='Expected the error to generate a warning, if it is initialized with an '
                    'errorUri containing an invalid character.')
            self.assertTrue(issubclass(caughtWarnings[0].category, RuntimeWarning),
                            msg='Expected the error to generate a RuntimeWarning.')
            self.assertIn(
                'Invalid character {space!r} in parameter'.format(space=' '),
                str(caughtWarnings[0].message),
                msg='Expected the warning to explain, that a character is invalid')
        self.assertEqual(self._VALID_ERROR_URI + '?', error.errorUri,
                         msg='Expected the error to escape spaces in the errorUri parameter.')

    def testGenerate(self):
        """ Test that the error modifies the request according to the OAuth2 specification. """
        additionalArguments = {None: None, 'description': self._VALID_DESCRIPTION,
                               'errorUri': self._VALID_ERROR_URI, 'scope': self._VALID_SCOPE}
        arguments = {}
        for additionalArgName, additionalArg in additionalArguments.items():
            if additionalArgName is not None:
                arguments[additionalArgName] = additionalArg
            error = self._initializeError(self._VALID_CODE, self._VALID_NAME, **arguments)
            request = MockRequest('GET', 'oauth2')
            resultBytes = error.generate(request)
            self.assertValidErrorResponse(
                request, resultBytes, self._VALID_CODE, name=self._VALID_NAME, **arguments)
        arguments = {}
        additionalArguments = {'addWwwAuthenticateHeader': True, 'errorUri': self._VALID_ERROR_URI,
                               'scope': None, 'authScheme': 'Test'}
        for parameterName, parameterValue in additionalArguments.items():
            arguments[parameterName] = parameterValue
            error = self._initializeError(self._VALID_CODE, self._VALID_NAME, **arguments)
            request = MockRequest('GET', 'oauth2')
            resultBytes = error.generate(request)
            self.assertValidErrorResponse(
                request, resultBytes, self._VALID_CODE, self._VALID_NAME, **arguments)

    def testCanBeThrown(self):
        """ Test that the error can be thrown as a normal Python exception. """
        error = self._initializeError(
            self._VALID_CODE, self._VALID_NAME, description=self._VALID_DESCRIPTION,
            errorUri=self._VALID_ERROR_URI, scope=self._VALID_SCOPE)
        try:
            raise error
        except error.__class__ as caughtError:
            self.assertIs(caughtError, error, msg='Expected to be able to raise the {name}.'
                          .format(name=error.__class__.__name__))
            self.assertEqual(self._VALID_DESCRIPTION, str(error),
                             msg='Expected the error to have the description as the error message.')
        error = self._initializeError(self._VALID_CODE, self._VALID_NAME, )
        try:
            raise error
        except error.__class__ as caughtError:
            self.assertIs(caughtError, error, msg='Expected to be able to raise the {name}.'
                          .format(name=error.__class__.__name__))
            self.assertEqual(
                self._VALID_NAME, str(error), msg='Expected the error to have the name as the '
                                                  'error message in absence of a description.')


class AuthorizationErrorTest(OAuth2ErrorTest):
    """ Test for AuthorizationErrors. """

    @staticmethod
    def _initializeError(code, name, **kwargs):
        return AuthorizationError(code, name, **kwargs)

    def assertRedirectsTo(self, request, redirectUri, msg):
        """
        Assert that the request redirects to the given uri.

        :param request: The request that should redirect.
        :param redirectUri: The uri where the request should redirect to.
        :param msg: The assertion message.
        """
        self.assertEqual(FOUND, request.responseCode,
                         msg=msg + ': Expected the auth resource to redirect the resource owner.')
        redirectUrl = request.getResponseHeader(b'location')
        self.assertIsNotNone(
            redirectUrl, msg=msg + ': Expected the auth resource to redirect the resource owner.')
        parsedUrl = urlparse(redirectUrl.decode('utf-8'))
        parsedUri = urlparse(redirectUri)
        self.assertTrue(
            parsedUrl.scheme == parsedUri.scheme and parsedUrl.netloc == parsedUri.netloc and
            parsedUrl.path == parsedUri.path and parsedUrl.params == parsedUri.params,
            msg='{msg}: The auth token endpoint did not redirect '
                'the resource owner to the expected url: '
                '{expected} <> {actual}'.format(msg=msg, expected=redirectUri, actual=redirectUrl))
        self.assertListEqual(sorted(parsedUri.query.split('&')), sorted(parsedUrl.query.split('&')),
                             msg=msg + ': Expected the redirect uri to contain the expected query.')
        self.assertListEqual(
            sorted(parsedUri.fragment.split('&')), sorted(parsedUrl.fragment.split('&')),
            msg=msg + ': Expected the redirect uri to contain the expected fragment.')

    def assertErrorRedirectsTo(self, error, baseRedirectUri, errorInFragment, targetUrl, msg):
        """
        Assert that the error will redirect a request to the expected target url.

        :param error: The error to test.
        :param baseRedirectUri: The redirect url that will be given to the error.
        :param errorInFragment: Whether the error should add the error information
                                to the fragment or to the query.
        :param targetUrl: The exact target url the error should redirect to.
        :param msg: An assertion message for failures.
        """
        request = MockRequest('GET', 'test')
        self.assertEqual(NOT_DONE_YET, error.generate(
            request, redirectUri=baseRedirectUri, errorInFragment=errorInFragment),
                         msg=msg + ': Expected the error to return NOT_DONE_YET.')
        self.assertRedirectsTo(request, targetUrl, msg=msg)

    def testValidatesStateParameter(self):
        """ Test that the error validates that the state parameter is a byte string. """
        for validValue in [b'state', b'\xFF\xFF', b'state\xFF\xFF', b'']:
            try:
                self._initializeError(self._VALID_CODE, self._VALID_NAME, state=validValue)
            except TypeError as error:
                self.fail('Expected the error to accept a valid value {value!r} as a state, '
                          'but got: {msg}'.format(value=validValue, msg=error))
        invalidString = 'normalString' if isinstance(u'', str) else u'unicodeString'
        for invalidValue in [True, invalidString, [], {}, object(), 1, 1.5]:
            self.assertRaises(TypeError, self._initializeError,
                              self._VALID_CODE, self._VALID_NAME, state=invalidValue)

    def testRedirectsIfItGetsRedirectUri(self):
        """ Test that the error redirects the request instead of returning the error. """
        self.assertErrorRedirectsTo(
            error=self._initializeError(self._VALID_CODE, 'name'),
            baseRedirectUri='https://redirect.com',
            errorInFragment=False,
            targetUrl='https://redirect.com?error=name',
            msg='Expected the error to redirect with the error info in the query')
        self.assertErrorRedirectsTo(
            error=self._initializeError(self._VALID_CODE, 'name'),
            baseRedirectUri='https://name:password@redirect.com:443',
            errorInFragment=False,
            targetUrl='https://name:password@redirect.com:443?error=name',
            msg='Expected the error to redirect with the error info in the query')
        self.assertErrorRedirectsTo(
            error=self._initializeError(self._VALID_CODE, 'name'),
            baseRedirectUri='https://redirect.com',
            errorInFragment=True,
            targetUrl='https://redirect.com#error=name',
            msg='Expected the error to redirect with the error info in the fragment')

    def testRetainsOriginalQuery(self):
        """ Test that the error retains the original query of the redirect uri. """
        self.assertErrorRedirectsTo(
            error=self._initializeError(self._VALID_CODE, 'name'),
            baseRedirectUri='https://redirect.com?param=test',
            errorInFragment=False,
            targetUrl='https://redirect.com?param=test&error=name',
            msg='Expected the error to retain query parameters from the original redirect uri')

    def testOverwritesOriginalQueryParameter(self):
        """ Test that the error overrides existing query parameters with the same name. """
        self.assertErrorRedirectsTo(
            error=self._initializeError(self._VALID_CODE, 'name'),
            baseRedirectUri='https://redirect.com?param=test&error=none',
            errorInFragment=False,
            targetUrl='https://redirect.com?param=test&error=name',
            msg='Expected the error to overwrite existing query parameters '
                'from the original redirect uri')
        self.assertErrorRedirectsTo(
            error=self._initializeError(self._VALID_CODE, 'name'),
            baseRedirectUri='https://redirect.com?param=test&error=none',
            errorInFragment=True,
            targetUrl='https://redirect.com?param=test&error=none#error=name',
            msg='Expected the error not to overwrite existing query parameters '
                'from the original redirect uri if the error information is send in the fragment.')

    def testRedirectsWithState(self):
        """ Test that the state will be send in the redirect uri. """
        self.assertErrorRedirectsTo(
            error=self._initializeError(self._VALID_CODE, 'name', state=b'stateXYZ'),
            baseRedirectUri='https://redirect.com',
            errorInFragment=False,
            targetUrl='https://redirect.com?error=name&state=stateXYZ',
            msg='Expected the error to include the state in the redirect url')
        self.assertErrorRedirectsTo(
            error=self._initializeError(self._VALID_CODE, 'name', state=b'state\xFF\xFFXYZ'),
            baseRedirectUri='https://redirect.com',
            errorInFragment=False,
            targetUrl='https://redirect.com?error=name&state=state%FF%FFXYZ',
            msg='Expected the error to include the state in the redirect url '
                'even if it includes invalid utf-8')

    def testRedirectsWithErrorInfo(self):
        """ Test that all optional additional information will be send with the redirect uri. """
        nameMapping = {'description': 'error_description', 'errorUri': 'error_uri'}
        additionalErrorInfo = {'description': self._VALID_DESCRIPTION,
                               'errorUri': self._VALID_ERROR_URI, 'scope': self._VALID_SCOPE[0]}
        for i in range(1, 2 ** len(additionalErrorInfo)):
            info = {key: value for j, (key, value) in enumerate(additionalErrorInfo.items())
                    if (i >> j) & 0b1}
            urlInfo = {nameMapping.get(key, key): value for key, value in info.items()}
            self.assertErrorRedirectsTo(
                error=self._initializeError(self._VALID_CODE, 'name', **info),
                baseRedirectUri='https://redirect.com',
                errorInFragment=False,
                targetUrl='https://redirect.com?error=name&' + urlencode(urlInfo),
                msg='Expected the error to include all error information in the redirect url query')
            self.assertErrorRedirectsTo(
                error=self._initializeError(self._VALID_CODE, 'name', **info),
                baseRedirectUri='https://redirect.com',
                errorInFragment=True,
                targetUrl='https://redirect.com#error=name&' + urlencode(urlInfo),
                msg='Expected the error to include all error information '
                    'in the redirect url fragment')

""" Tests for all OAuth2 errors. """
import json
import warnings

try:
    from __builtin__ import unichr as _unichr
except ImportError:
    _unichr = chr

try:
    from urlparse import urlparse
    from urllib import urlencode
except ImportError:
    # noinspection PyUnresolvedReferences
    from urllib.parse import urlparse, urlencode

from twisted.web.server import NOT_DONE_YET
from twisted.web.http import ACCEPTED, UNAUTHORIZED, FOUND
from txoauth2.errors import OAuth2Error, AuthorizationError

from tests import TwistedTestCase, MockRequest


class OAuth2ErrorTest(TwistedTestCase):
    """ Test for OAuth2Errors. """
    _VALID_CODE = ACCEPTED
    _VALID_NAME = 'Error name'
    _VALID_DESCRIPTION = 'Error description'
    _VALID_ERROR_URI = 'error:uri'
    _VALID_SCOPE = ['scope']

    @staticmethod
    def _initializeError(code, name, **kwargs):
        """
        Initialize a new error instance that should be tested.

        :param code: The return code of the error.
        :param name: The name of the error.
        :param kwargs: Additional arguments for the error.
        :return: The new initialized error.
        """
        return OAuth2Error(code, name, **kwargs)

    def assertValidErrorResponse(
            self, request, resultBytes, code=None, name=None, description=None, errorUri=None,
            scope=None, addWwwAuthenticateHeader=False, authScheme='Bearer'):
        """
        Assert that the response generated by the error matches the expected error response.

        :param request: The request that was modified by the error.
        :param resultBytes: The result that was returned by the error.
        :param code: The expected HTTP return code.
        :param name: The expected error name.
        :param description: The expected error description.
        :param errorUri: The expected error uri.
        :param scope: The expected scope.
        :param addWwwAuthenticateHeader: Whether or not the error should have added a
                                         WWW-Authenticate response header.
        :param authScheme: The expected authScheme in the response WWW-Authenticate header.
        """
        parameter = {'error': name, 'error_description': description,
                     'error_uri': errorUri, 'scope': scope}
        self.assertEqual(
            code, request.responseCode, msg='Expected the error to set the response code.')
        self.assertEqual(
            'application/json;charset=UTF-8', request.getResponseHeader('Content-Type'),
            msg='Expected the error to indicate that the response is utf-8 encoded json.')
        self.assertEqual(
            'no-store', request.getResponseHeader('Cache-Control'),
            msg='Expected the error to indicate that the response should not be cached.')
        self.assertEqual(
            'no-cache', request.getResponseHeader('Pragma'),
            msg='Expected the error to indicate that the response should not be cached.')
        if addWwwAuthenticateHeader:
            self.assertIsNotNone(
                request.getResponseHeader('WWW-Authenticate'),
                msg='Expected the error to include a WWW-Authenticate header.')
            requestAuthScheme, params = request.getResponseHeader('WWW-Authenticate').split(' ', 1)
            self.assertEqual(authScheme, requestAuthScheme,
                             msg='Expected the error to use the configured auth scheme '
                                 'in the WWW-Authenticate header.')
            authenticateParameter = dict(item.split('=', 1) for item in params.split(','))
            for parameterName, value in parameter.items():
                if value is None:
                    self.assertNotIn(parameterName, authenticateParameter,
                                     msg='Expected the error not to add the {name} parameter to '
                                         'the WWW-Authenticate header.'.format(name=parameterName))
                    continue
                self.assertIn(parameterName, authenticateParameter,
                              msg='Expected the error to add the {name} parameter '
                                  'to the WWW-Authenticate header.'.format(name=parameterName))
                self.assertEqual(
                    '"' + value + '"', authenticateParameter.pop(parameterName),
                    msg='Expected the error to include the correct value for the parameter {name} '
                        'in the WWW-Authenticate header.'.format(name=parameterName))
            self.assertIn('realm', authenticateParameter,
                          msg='Expected the error to add the realm to the WWW-Authenticate header.')
            self.assertEqual(
                '"{realm}"'.format(realm=request.prePathURL().decode('utf-8')),
                authenticateParameter.pop('realm'),
                msg='Expected the error to include the correct realm '
                    'in the WWW-Authenticate header.')
            self.assertDictEqual(
                authenticateParameter, {}, msg='Expected the error not to return any additional '
                                               'parameters in the WWW-Authenticate header.')
        else:
            self.assertIsNone(
                request.getResponseHeader('WWW-Authenticate'),
                msg='Expected the error not to include a WWW-Authenticate header.')
        self.assertIsInstance(
            resultBytes, bytes, message='Expected the error to generate bytes.')
        result = json.loads(resultBytes.decode('utf-8'))
        self.assertIsInstance(result, dict, message='Expected the error return a json object.')
        parameter['scope'] = None
        for parameterName, value in parameter.items():
            if value is None:
                self.assertNotIn(parameterName, result,
                                 msg='Expected the error not to add the {name} '
                                     'parameter to the response.'.format(name=parameterName))
                continue
            self.assertIn(parameterName, result,
                          msg='Expected the error to add the {name} parameter '
                              'to the response.'.format(name=parameterName))
            self.assertEqual(value, result.pop(parameterName),
                             msg='Expected the error to return the correct value '
                                 'for the parameter {name}.'.format(name=parameterName))
        self.assertDictEqual(
            result, {}, msg='Expected the error not to return any additional parameters.')

    def testInitialization(self):
        """ Test that the error can be initialized correctly and checks the argument types. """
        authScheme = 'Test'
        error = self._initializeError(
            self._VALID_CODE, self._VALID_NAME, description=self._VALID_DESCRIPTION,
            errorUri=self._VALID_ERROR_URI, scope=self._VALID_SCOPE,
            addWwwAuthenticateHeader=True, authScheme=authScheme)
        self.assertEqual(self._VALID_CODE, error.code,
                         msg='Expected the {error} to use the code that it has been '
                             'initialized with.'.format(error=error.__class__.__name__))
        self.assertEqual(self._VALID_NAME, error.name,
                         msg='Expected the {error} to use the name that it has been '
                             'initialized with.'.format(error=error.__class__.__name__))
        self.assertEqual(self._VALID_DESCRIPTION, error.description,
                         msg='Expected the {error} to use the description that it has been '
                             'initialized with.'.format(error=error.__class__.__name__))
        self.assertEqual(self._VALID_ERROR_URI, error.errorUri,
                         msg='Expected the {error} to use the error uri that it has been '
                             'initialized with.'.format(error=error.__class__.__name__))
        self.assertListEqual(error.scope, self._VALID_SCOPE,
                             msg='Expected the {error} to use the scope that it has been '
                                 'initialized with.'.format(error=error.__class__.__name__))
        self.assertTrue(error._addWwwAuthenticateHeader,  # pylint: disable=protected-access
                        msg='Expected the {error} to enable the WWW-Authenticate header.'.format(
                            error=error.__class__.__name__))
        self.assertEqual(authScheme, error._authScheme,  # pylint: disable=protected-access
                         msg='Expected the {error} to use the auth scheme that it has been '
                             'initialized with.'.format(error=error.__class__.__name__))
        error = self._initializeError(UNAUTHORIZED, self._VALID_NAME, scope=self._VALID_SCOPE[0])
        self.assertTrue(error._addWwwAuthenticateHeader,  # pylint: disable=protected-access
                        msg='Expected the {error} to enable the WWW-Authenticate header if the '
                            'code is 401 (UNAUTHORIZED).'.format(error=error.__class__.__name__))
        self.assertListEqual(error.scope, self._VALID_SCOPE,
                             msg='Expected the {error} to accept a single scope item.'
                             .format(error=error.__class__.__name__))
        error = self._initializeError(
            self._VALID_CODE, self._VALID_NAME, scope=[], addWwwAuthenticateHeader=False)
        self.assertFalse(error._addWwwAuthenticateHeader,  # pylint: disable=protected-access
                         msg='Expected the {error} not to enable the WWW-Authenticate header if it '
                             'is explicitly disabled.'.format(error=error.__class__.__name__))
        self.assertIsNone(error.scope, msg='Expected the {error} to treat an empty scope list as '
                                           'no scope.'.format(error=error.__class__.__name__))

    def testValidatesParameters(self):
        """ Test that invalid parameter values raise the appropriate exception. """
        for invalidCode in [None, '1', b'1', u'1', [], {}, object(), 2.0]:
            self.assertRaises(TypeError, self._initializeError, invalidCode, self._VALID_NAME)
        invalidValues = [1, [], {}, object(), 2.0]
        if not isinstance(b'', str):
            invalidValues.append(b'byteString')
        for invalidValue in invalidValues:
            self.assertRaises(TypeError, self._initializeError, self._VALID_CODE, invalidValue)
            self.assertRaises(TypeError, self._initializeError, self._VALID_CODE,
                              self._VALID_NAME, invalidValue)
            self.assertRaises(TypeError, self._initializeError, self._VALID_CODE,
                              self._VALID_NAME, errorUri=invalidValue)
            if not isinstance(invalidValue, list):
                self.assertRaises(TypeError, self._initializeError, self._VALID_CODE,
                                  self._VALID_NAME, scope=invalidValue)
            self.assertRaises(TypeError, self._initializeError, self._VALID_CODE,
                              self._VALID_NAME, scope=['valid', invalidValue])
            self.assertRaises(TypeError, self._initializeError, self._VALID_CODE,
                              self._VALID_NAME, authScheme=invalidValue)
        self.assertRaises(TypeError, self._initializeError, self._VALID_CODE, None,
                          self._VALID_DESCRIPTION, self._VALID_ERROR_URI)
        self.assertRaises(TypeError, self._initializeError, self._VALID_CODE,
                          self._VALID_NAME, addWwwAuthenticateHeader=object())
        invalidCharacters = [chr(char) for char in range(0x20)] + [
            _unichr(char) for char in range(0x7F, 0x120)] + ['"', '\\', '\xFF']
        for invalidCharacter in invalidCharacters:
            with warnings.catch_warnings(record=True) as caughtWarnings:
                warnings.simplefilter('always')
                error = self._initializeError(
                    self._VALID_CODE, self._VALID_NAME + invalidCharacter,
                    description=self._VALID_DESCRIPTION + invalidCharacter,
                    errorUri=self._VALID_ERROR_URI + invalidCharacter,
                    scope=['a' + invalidCharacter, 'b' + invalidCharacter])
                self.assertEqual(
                    4, len(caughtWarnings),
                    msg='Expected the {error} to generate 4 warnings for 4 invalid characters, if '
                        'it is initialized with a name, an error uri and two scopes containing an '
                        'invalid character.'.format(error=error.__class__.__name__))
                for warning in caughtWarnings:
                    self.assertTrue(issubclass(warning.category, RuntimeWarning),
                                    msg='Expected the {error} to generate a RuntimeWarning.'
                                    .format(error=error.__class__.__name__))
                    self.assertIn(
                        'Invalid character {char!r} in parameter'.format(char=invalidCharacter),
                        str(warning.message),
                        msg='Expected the warning to explain, that a character is invalid: '
                            '{warning}'.format(warning=warning.message))
            self.assertEqual(self._VALID_NAME + '?', error.name,
                             msg='Expected the error to escape the name parameter.')
            self.assertEqual(self._VALID_DESCRIPTION + '?', error.description,
                             msg='Expected the error to escape the description parameter.')
            self.assertEqual(self._VALID_ERROR_URI + '?', error.errorUri,
                             msg='Expected the error to escape the errorUri parameter.')
            self.assertListEqual(['a?', 'b?'], error.scope,
                                 msg='Expected the error to escape the scope parameter.')
        with warnings.catch_warnings(record=True) as caughtWarnings:
            warnings.simplefilter('always')
            error = self._initializeError(
                self._VALID_CODE, self._VALID_NAME,
                description=self._VALID_DESCRIPTION, errorUri=self._VALID_ERROR_URI + ' ')
            self.assertEqual(
                1, len(caughtWarnings),
                msg='Expected the error to generate a warning, if it is initialized with an '
                    'errorUri containing an invalid character.')
            self.assertTrue(issubclass(caughtWarnings[0].category, RuntimeWarning),
                            msg='Expected the error to generate a RuntimeWarning.')
            self.assertIn(
                'Invalid character {space!r} in parameter'.format(space=' '),
                str(caughtWarnings[0].message),
                msg='Expected the warning to explain, that a character is invalid')
        self.assertEqual(self._VALID_ERROR_URI + '?', error.errorUri,
                         msg='Expected the error to escape spaces in the errorUri parameter.')

    def testGenerate(self):
        """ Test that the error modifies the request according to the OAuth2 specification. """
        additionalArguments = {None: None, 'description': self._VALID_DESCRIPTION,
                               'errorUri': self._VALID_ERROR_URI, 'scope': self._VALID_SCOPE}
        arguments = {}
        for additionalArgName, additionalArg in additionalArguments.items():
            if additionalArgName is not None:
                arguments[additionalArgName] = additionalArg
            error = self._initializeError(self._VALID_CODE, self._VALID_NAME, **arguments)
            request = MockRequest('GET', 'oauth2')
            resultBytes = error.generate(request)
            self.assertValidErrorResponse(
                request, resultBytes, self._VALID_CODE, name=self._VALID_NAME, **arguments)
        arguments = {}
        additionalArguments = {'addWwwAuthenticateHeader': True, 'errorUri': self._VALID_ERROR_URI,
                               'scope': None, 'authScheme': 'Test'}
        for parameterName, parameterValue in additionalArguments.items():
            arguments[parameterName] = parameterValue
            error = self._initializeError(self._VALID_CODE, self._VALID_NAME, **arguments)
            request = MockRequest('GET', 'oauth2')
            resultBytes = error.generate(request)
            self.assertValidErrorResponse(
                request, resultBytes, self._VALID_CODE, self._VALID_NAME, **arguments)

    def testCanBeThrown(self):
        """ Test that the error can be thrown as a normal Python exception. """
        error = self._initializeError(
            self._VALID_CODE, self._VALID_NAME, description=self._VALID_DESCRIPTION,
            errorUri=self._VALID_ERROR_URI, scope=self._VALID_SCOPE)
        try:
            raise error
        except error.__class__ as caughtError:
            self.assertIs(caughtError, error, msg='Expected to be able to raise the {name}.'
                          .format(name=error.__class__.__name__))
            self.assertEqual(self._VALID_DESCRIPTION, str(error),
                             msg='Expected the error to have the description as the error message.')
        error = self._initializeError(self._VALID_CODE, self._VALID_NAME, )
        try:
            raise error
        except error.__class__ as caughtError:
            self.assertIs(caughtError, error, msg='Expected to be able to raise the {name}.'
                          .format(name=error.__class__.__name__))
            self.assertEqual(
                self._VALID_NAME, str(error), msg='Expected the error to have the name as the '
                                                  'error message in absence of a description.')


class AuthorizationErrorTest(OAuth2ErrorTest):
    """ Test for AuthorizationErrors. """

    @staticmethod
    def _initializeError(code, name, **kwargs):
        return AuthorizationError(code, name, **kwargs)

    def assertRedirectsTo(self, request, redirectUri, msg):
        """
        Assert that the request redirects to the given uri.

        :param request: The request that should redirect.
        :param redirectUri: The uri where the request should redirect to.
        :param msg: The assertion message.
        """
        self.assertEqual(FOUND, request.responseCode,
                         msg=msg + ': Expected the auth resource to redirect the resource owner.')
        redirectUrl = request.getResponseHeader(b'location')
        self.assertIsNotNone(
            redirectUrl, msg=msg + ': Expected the auth resource to redirect the resource owner.')
        parsedUrl = urlparse(redirectUrl.decode('utf-8'))
        parsedUri = urlparse(redirectUri)
        self.assertTrue(
            parsedUrl.scheme == parsedUri.scheme and parsedUrl.netloc == parsedUri.netloc and
            parsedUrl.path == parsedUri.path and parsedUrl.params == parsedUri.params,
            msg='{msg}: The auth token endpoint did not redirect '
                'the resource owner to the expected url: '
                '{expected} <> {actual}'.format(msg=msg, expected=redirectUri, actual=redirectUrl))
        self.assertListEqual(sorted(parsedUri.query.split('&')), sorted(parsedUrl.query.split('&')),
                             msg=msg + ': Expected the redirect uri to contain the expected query.')
        self.assertListEqual(
            sorted(parsedUri.fragment.split('&')), sorted(parsedUrl.fragment.split('&')),
            msg=msg + ': Expected the redirect uri to contain the expected fragment.')

    def assertErrorRedirectsTo(self, error, baseRedirectUri, errorInFragment, targetUrl, msg):
        """
        Assert that the error will redirect a request to the expected target url.

        :param error: The error to test.
        :param baseRedirectUri: The redirect url that will be given to the error.
        :param errorInFragment: Whether the error should add the error information
                                to the fragment or to the query.
        :param targetUrl: The exact target url the error should redirect to.
        :param msg: An assertion message for failures.
        """
        request = MockRequest('GET', 'test')
        self.assertEqual(NOT_DONE_YET, error.generate(
            request, redirectUri=baseRedirectUri, errorInFragment=errorInFragment),
                         msg=msg + ': Expected the error to return NOT_DONE_YET.')
        self.assertRedirectsTo(request, targetUrl, msg=msg)

    def testValidatesStateParameter(self):
        """ Test that the error validates that the state parameter is a byte string. """
        for validValue in [b'state', b'\xFF\xFF', b'state\xFF\xFF', b'']:
            try:
                self._initializeError(self._VALID_CODE, self._VALID_NAME, state=validValue)
            except TypeError as error:
                self.fail('Expected the error to accept a valid value {value!r} as a state, '
                          'but got: {msg}'.format(value=validValue, msg=error))
        invalidString = 'normalString' if isinstance(u'', str) else u'unicodeString'
        for invalidValue in [True, invalidString, [], {}, object(), 1, 1.5]:
            self.assertRaises(TypeError, self._initializeError,
                              self._VALID_CODE, self._VALID_NAME, state=invalidValue)

    def testRedirectsIfItGetsRedirectUri(self):
        """ Test that the error redirects the request instead of returning the error. """
        self.assertErrorRedirectsTo(
            error=self._initializeError(self._VALID_CODE, 'name'),
            baseRedirectUri='https://redirect.com',
            errorInFragment=False,
            targetUrl='https://redirect.com?error=name',
            msg='Expected the error to redirect with the error info in the query')
        self.assertErrorRedirectsTo(
            error=self._initializeError(self._VALID_CODE, 'name'),
            baseRedirectUri='https://name:password@redirect.com:443',
            errorInFragment=False,
            targetUrl='https://name:password@redirect.com:443?error=name',
            msg='Expected the error to redirect with the error info in the query')
        self.assertErrorRedirectsTo(
            error=self._initializeError(self._VALID_CODE, 'name'),
            baseRedirectUri='https://redirect.com',
            errorInFragment=True,
            targetUrl='https://redirect.com#error=name',
            msg='Expected the error to redirect with the error info in the fragment')

    def testRetainsOriginalQuery(self):
        """ Test that the error retains the original query of the redirect uri. """
        self.assertErrorRedirectsTo(
            error=self._initializeError(self._VALID_CODE, 'name'),
            baseRedirectUri='https://redirect.com?param=test',
            errorInFragment=False,
            targetUrl='https://redirect.com?param=test&error=name',
            msg='Expected the error to retain query parameters from the original redirect uri')

    def testOverwritesOriginalQueryParameter(self):
        """ Test that the error overrides existing query parameters with the same name. """
        self.assertErrorRedirectsTo(
            error=self._initializeError(self._VALID_CODE, 'name'),
            baseRedirectUri='https://redirect.com?param=test&error=none',
            errorInFragment=False,
            targetUrl='https://redirect.com?param=test&error=name',
            msg='Expected the error to overwrite existing query parameters '
                'from the original redirect uri')
        self.assertErrorRedirectsTo(
            error=self._initializeError(self._VALID_CODE, 'name'),
            baseRedirectUri='https://redirect.com?param=test&error=none',
            errorInFragment=True,
            targetUrl='https://redirect.com?param=test&error=none#error=name',
            msg='Expected the error not to overwrite existing query parameters '
                'from the original redirect uri if the error information is send in the fragment.')

    def testRedirectsWithState(self):
        """ Test that the state will be send in the redirect uri. """
        self.assertErrorRedirectsTo(
            error=self._initializeError(self._VALID_CODE, 'name', state=b'stateXYZ'),
            baseRedirectUri='https://redirect.com',
            errorInFragment=False,
            targetUrl='https://redirect.com?error=name&state=stateXYZ',
            msg='Expected the error to include the state in the redirect url')
        self.assertErrorRedirectsTo(
            error=self._initializeError(self._VALID_CODE, 'name', state=b'state\xFF\xFFXYZ'),
            baseRedirectUri='https://redirect.com',
            errorInFragment=False,
            targetUrl='https://redirect.com?error=name&state=state%FF%FFXYZ',
            msg='Expected the error to include the state in the redirect url '
                'even if it includes invalid utf-8')

    def testRedirectsWithErrorInfo(self):
        """ Test that all optional additional information will be send with the redirect uri. """
        nameMapping = {'description': 'error_description', 'errorUri': 'error_uri'}
        additionalErrorInfo = {'description': self._VALID_DESCRIPTION,
                               'errorUri': self._VALID_ERROR_URI, 'scope': self._VALID_SCOPE[0]}
        for i in range(1, 2 ** len(additionalErrorInfo)):
            info = {key: value for j, (key, value) in enumerate(additionalErrorInfo.items())
                    if (i >> j) & 0b1}
            urlInfo = {nameMapping.get(key, key): value for key, value in info.items()}
            self.assertErrorRedirectsTo(
                error=self._initializeError(self._VALID_CODE, 'name', **info),
                baseRedirectUri='https://redirect.com',
                errorInFragment=False,
                targetUrl='https://redirect.com?error=name&' + urlencode(urlInfo),
                msg='Expected the error to include all error information in the redirect url query')
            self.assertErrorRedirectsTo(
                error=self._initializeError(self._VALID_CODE, 'name', **info),
                baseRedirectUri='https://redirect.com',
                errorInFragment=True,
                targetUrl='https://redirect.com#error=name&' + urlencode(urlInfo),
                msg='Expected the error to include all error information '
                    'in the redirect url fragment')

Tests for all OAuth2 errors. # noinspection PyUnresolvedReferences Test for OAuth2Errors. Initialize a new error instance that should be tested.          :param code: The return code of the error.         :param name: The name of the error.         :param kwargs: Additional arguments for the error.         :return: The new initialized error. Assert that the response generated by the error matches the expected error response.          :param request: The request that was modified by the error.         :param resultBytes: The result that was returned by the error.         :param code: The expected HTTP return code.         :param name: The expected error name.         :param description: The expected error description.         :param errorUri: The expected error uri.         :param scope: The expected scope.         :param addWwwAuthenticateHeader: Whether or not the error should have added a                                          WWW-Authenticate response header.         :param authScheme: The expected authScheme in the response WWW-Authenticate header. Test that the error can be initialized correctly and checks the argument types. # pylint: disable=protected-access # pylint: disable=protected-access # pylint: disable=protected-access # pylint: disable=protected-access Test that invalid parameter values raise the appropriate exception. Test that the error modifies the request according to the OAuth2 specification. Test that the error can be thrown as a normal Python exception. Test for AuthorizationErrors. Assert that the request redirects to the given uri.          :param request: The request that should redirect.         :param redirectUri: The uri where the request should redirect to.         :param msg: The assertion message. Assert that the error will redirect a request to the expected target url.          :param error: The error to test.         :param baseRedirectUri: The redirect url that will be given to the error.         :param errorInFragment: Whether the error should add the error information                                 to the fragment or to the query.         :param targetUrl: The exact target url the error should redirect to.         :param msg: An assertion message for failures. Test that the error validates that the state parameter is a byte string. Test that the error redirects the request instead of returning the error. #error=name', Test that the error retains the original query of the redirect uri. Test that the error overrides existing query parameters with the same name. #error=name', Test that the state will be send in the redirect uri. Test that all optional additional information will be send with the redirect uri. #error=name&' + urlencode(urlInfo),