# -*- coding: utf-8 -*-

from __future__ import unicode_literals

import datetime
import json

import mock
import pytest

from oauthlib.oauth2 import InvalidRequestFatalError
from oauthlib.common import Request as OAuthRequest
from pyramid import httpexceptions

from h._compat import urlparse
from h.exceptions import OAuthTokenError
from h.models.auth_client import ResponseType
from h.services.auth_token import auth_token_service_factory
from h.services.oauth_provider import OAuthProviderService
from h.services.oauth_validator import DEFAULT_SCOPES
from h.services.user import user_service_factory
from h.util.datetime import utc_iso8601
from h.views import api_auth as views


@pytest.mark.usefixtures('routes', 'oauth_provider', 'user_svc')
class TestOAuthAuthorizeController(object):
    @pytest.mark.usefixtures('authenticated_user')
    @pytest.mark.parametrize('view_name', ['get', 'get_web_message'])
    def test_get_validates_request(self, controller, pyramid_request, view_name):
        view = getattr(controller, view_name)
        view()

        controller.oauth.validate_authorization_request.assert_called_once_with(
            pyramid_request.url)

    @pytest.mark.parametrize('view_name', ['get', 'get_web_message'])
    def test_get_raises_for_invalid_request(self, controller, view_name):
        controller.oauth.validate_authorization_request.side_effect = InvalidRequestFatalError('boom!')

        with pytest.raises(InvalidRequestFatalError) as exc:
            view = getattr(controller, view_name)
            view()

        assert exc.value.description == 'boom!'

    @pytest.mark.parametrize('view_name', ['get', 'get_web_message'])
    def test_get_redirects_to_login_when_not_authenticated(self, controller, pyramid_request, view_name):
        with pytest.raises(httpexceptions.HTTPFound) as exc:
            view = getattr(controller, view_name)
            view()

        parsed_url = urlparse.urlparse(exc.value.location)
        assert parsed_url.path == '/login'
        assert urlparse.parse_qs(parsed_url.query) == {'next': [pyramid_request.url],
                                                       'for_oauth': ['True']}

    @pytest.mark.parametrize('response_mode,view_name', [
        (None, 'get'),
        ('web_message', 'get_web_message'),
    ])
    def test_get_returns_expected_context(self, controller, auth_client, authenticated_user, oauth_request, response_mode, view_name):
        oauth_request.response_mode = response_mode

        view = getattr(controller, view_name)
        assert view() == {
            'client_id': auth_client.id,
            'client_name': auth_client.name,
            'response_mode': response_mode,
            'response_type': auth_client.response_type.value,
            'state': 'foobar',
            'username': authenticated_user.username,
        }

    @pytest.mark.parametrize('view_name', ['get', 'get_web_message'])
    def test_get_creates_authorization_response_for_trusted_clients(self, controller, auth_client, authenticated_user, pyramid_request, view_name):
        auth_client.trusted = True

        view = getattr(controller, view_name)
        view()

        controller.oauth.create_authorization_response.assert_called_once_with(
            pyramid_request.url,
            credentials={'user': authenticated_user},
            scopes=DEFAULT_SCOPES)

    def test_get_returns_redirect_immediately_for_trusted_clients(self, controller, auth_client, authenticated_user, pyramid_request):
        auth_client.trusted = True

        response = controller.get()
        expected = '{}?code=abcdef123456&state=foobar'.format(auth_client.redirect_uri)

        assert response.location == expected

    @pytest.mark.usefixtures('authenticated_user')
    def test_get_web_message_renders_template_for_trusted_clients(self, controller, auth_client):
        auth_client.trusted = True

        assert controller.request.override_renderer is None
        controller.get_web_message()
        assert controller.request.override_renderer == 'h:templates/oauth/authorize_web_message.html.jinja2'

    @pytest.mark.usefixtures('authenticated_user')
    def test_get_web_message_returns_context_for_trusted_clients(self, controller, auth_client):
        auth_client.trusted = True

        response = controller.get_web_message()

        assert response == {
            'code': 'abcdef123456',
            'origin': 'http://client.com',
            'state': 'foobar',
        }

    @pytest.mark.usefixtures('authenticated_user')
    def test_get_web_message_allows_empty_state_in_context_for_trusted_clients(self, controller, auth_client, oauth_provider):
        auth_client.trusted = True

        headers = {'Location': '{}?code=abcdef123456'.format(auth_client.redirect_uri)}
        oauth_provider.create_authorization_response.return_value = (headers, None, 302)

        response = controller.get_web_message()
        assert response['state'] is None

    @pytest.mark.parametrize('view_name', ['post', 'post_web_message'])
    def test_post_creates_authorization_response(self, controller, pyramid_request, authenticated_user, view_name):
        pyramid_request.url = 'http://example.com/auth?client_id=the-client-id' + \
                                                     '&response_type=code' + \
                                                     '&state=foobar' + \
                                                     '&scope=exploit'

        view = getattr(controller, view_name)
        view()

        controller.oauth.create_authorization_response.assert_called_once_with(
            pyramid_request.url,
            credentials={'user': authenticated_user},
            scopes=DEFAULT_SCOPES)

    @pytest.mark.usefixtures('authenticated_user')
    @pytest.mark.parametrize('view_name', ['post', 'post_web_message'])
    def test_post_raises_for_invalid_request(self, controller, view_name):
        controller.oauth.create_authorization_response.side_effect = InvalidRequestFatalError('boom!')

        with pytest.raises(InvalidRequestFatalError) as exc:
            view = getattr(controller, view_name)
            view()

        assert exc.value.description == 'boom!'

    def test_post_redirects_to_client(self, controller, auth_client):
        response = controller.post()
        expected = '{}?code=abcdef123456&state=foobar'.format(auth_client.redirect_uri)

        assert response.location == expected

    def test_post_web_message_returns_expected_context(self, controller, auth_client):
        response = controller.post_web_message()

        assert response == {
            'code': 'abcdef123456',
            'origin': 'http://client.com',
            'state': 'foobar',
        }

    def test_post_web_message_allows_empty_state_in_context(self, controller, auth_client, oauth_provider):
        auth_client.trusted = True

        headers = {'Location': '{}?code=abcdef123456'.format(auth_client.redirect_uri)}
        oauth_provider.create_authorization_response.return_value = (headers, None, 302)

        response = controller.post_web_message()
        assert response['state'] is None

    @pytest.fixture
    def controller(self, pyramid_request):
        pyramid_request.override_renderer = None
        return views.OAuthAuthorizeController(None, pyramid_request)

    @pytest.fixture
    def oauth_request(self):
        return OAuthRequest('/')

    @pytest.fixture
    def oauth_provider(self, pyramid_config, auth_client, oauth_request):
        svc = mock.create_autospec(OAuthProviderService, instance=True)

        scopes = ['annotation:read', 'annotation:write']
        credentials = {'client_id': auth_client.id, 'state': 'foobar', 'request': oauth_request}
        svc.validate_authorization_request.return_value = (scopes, credentials)

        headers = {'Location': '{}?code=abcdef123456&state=foobar'.format(auth_client.redirect_uri)}
        body = None
        status = 302
        svc.create_authorization_response.return_value = (headers, body, status)

        pyramid_config.register_service(svc, name='oauth_provider')

        return svc

    @pytest.fixture
    def auth_client(self, factories):
        return factories.AuthClient(name='Test Client',
                                    redirect_uri='http://client.com/auth/callback',
                                    response_type=ResponseType.code)

    @pytest.fixture
    def user_svc(self, pyramid_config, pyramid_request):
        svc = mock.Mock(spec_set=user_service_factory(None, pyramid_request))
        pyramid_config.register_service(svc, name='user')
        return svc

    @pytest.fixture
    def pyramid_request(self, pyramid_request):
        pyramid_request.url = 'http://example.com/auth?client_id=the-client-id&response_type=code&state=foobar'
        return pyramid_request

    @pytest.fixture
    def authenticated_user(self, factories, pyramid_config, user_svc):
        user = factories.User.build()
        pyramid_config.testing_securitypolicy(user.userid)

        def fake_fetch(userid):
            if userid == user.userid:
                return user
        user_svc.fetch.side_effect = fake_fetch

        return user

    @pytest.fixture
    def routes(self, pyramid_config):
        pyramid_config.add_route('login', '/login')


@pytest.mark.usefixtures('oauth_provider')
class TestOAuthAccessTokenController(object):
    def test_it_creates_token_response(self, pyramid_request, controller, oauth_provider):
        controller.post()
        oauth_provider.create_token_response.assert_called_once_with(
                pyramid_request.url, pyramid_request.method, pyramid_request.POST, pyramid_request.headers)

    def test_it_returns_correct_response_on_success(self, controller, oauth_provider):
        body = json.dumps({'access_token': 'the-access-token'})
        oauth_provider.create_token_response.return_value = ({}, body, 200)

        assert controller.post() == {'access_token': 'the-access-token'}

    def test_it_raises_when_error(self, controller, oauth_provider):
        body = json.dumps({'error': 'invalid_request'})
        oauth_provider.create_token_response.return_value = ({}, body, 400)

        with pytest.raises(httpexceptions.HTTPBadRequest) as exc:
            controller.post()

        assert exc.value.body == body

    @pytest.fixture
    def controller(self, pyramid_request):
        pyramid_request.method = 'POST'
        pyramid_request.POST['grant_type'] = 'authorization_code'
        pyramid_request.POST['code'] = 'the-authz-code'
        pyramid_request.headers = {'X-Test-ID': '1234'}
        return views.OAuthAccessTokenController(pyramid_request)

    @pytest.fixture
    def oauth_provider(self, pyramid_config):
        svc = mock.Mock(spec_set=['create_token_response'])
        svc.create_token_response.return_value = ({}, '{}', 200)
        pyramid_config.register_service(svc, name='oauth_provider')
        return svc


@pytest.mark.usefixtures('oauth_provider')
class TestOAuthRevocationController(object):
    def test_it_creates_revocation_response(self, pyramid_request, controller, oauth_provider):
        controller.post()
        oauth_provider.create_revocation_response.assert_called_once_with(
                pyramid_request.url, pyramid_request.method, pyramid_request.POST, pyramid_request.headers)

    def test_it_returns_empty_response_on_success(self, controller):
        response = controller.post()
        assert response == {}

    def test_it_raises_when_error(self, controller, oauth_provider):
        body = json.dumps({'error': 'invalid_request'})
        oauth_provider.create_revocation_response.return_value = ({}, body, 400)

        with pytest.raises(httpexceptions.HTTPBadRequest) as exc:
            controller.post()

        assert exc.value.body == body

    @pytest.fixture
    def controller(self, pyramid_request):
        pyramid_request.method = 'POST'
        pyramid_request.POST['token'] = 'the-token'
        pyramid_request.headers = {'X-Test-ID': '1234'}
        return views.OAuthRevocationController(pyramid_request)

    @pytest.fixture
    def oauth_provider(self, pyramid_config):
        svc = mock.Mock(spec_set=['create_revocation_response'])
        svc.create_revocation_response.return_value = ({}, '{}', 200)
        pyramid_config.register_service(svc, name='oauth_provider')
        return svc


class TestDebugToken(object):
    def test_it_raises_error_when_token_is_missing(self, pyramid_request):
        pyramid_request.auth_token = None

        with pytest.raises(OAuthTokenError) as exc:
            views.debug_token(pyramid_request)

        assert exc.value.type == 'missing_token'
        assert 'Bearer token is missing' in exc.value.message

    def test_it_raises_error_when_token_is_empty(self, pyramid_request):
        pyramid_request.auth_token = ''

        with pytest.raises(OAuthTokenError) as exc:
            views.debug_token(pyramid_request)

        assert exc.value.type == 'missing_token'
        assert 'Bearer token is missing' in exc.value.message

    def test_it_validates_token(self, pyramid_request, token_service):
        pyramid_request.auth_token = 'the-access-token'

        views.debug_token(pyramid_request)

        token_service.validate.assert_called_once_with('the-access-token')

    def test_it_raises_error_when_token_is_invalid(self, pyramid_request, token_service):
        pyramid_request.auth_token = 'the-token'
        token_service.validate.return_value = None

        with pytest.raises(OAuthTokenError) as exc:
            views.debug_token(pyramid_request)

        assert exc.value.type == 'missing_token'
        assert 'Bearer token does not exist or is expired' in exc.value.message

    def test_returns_debug_data_for_oauth_token(self, pyramid_request, token_service, oauth_token):
        pyramid_request.auth_token = oauth_token.value
        token_service.fetch.return_value = oauth_token

        result = views.debug_token(pyramid_request)

        assert result == {'userid': oauth_token.userid,
                          'client': {'id': oauth_token.authclient.id,
                                     'name': oauth_token.authclient.name},
                          'issued_at': utc_iso8601(oauth_token.created),
                          'expires_at': utc_iso8601(oauth_token.expires),
                          'expired': oauth_token.expired}

    def test_returns_debug_data_for_developer_token(self, pyramid_request, token_service, developer_token):
        pyramid_request.auth_token = developer_token.value
        token_service.fetch.return_value = developer_token

        result = views.debug_token(pyramid_request)

        assert result == {'userid': developer_token.userid,
                          'issued_at': utc_iso8601(developer_token.created),
                          'expires_at': None,
                          'expired': False}

    @pytest.fixture
    def token_service(self, pyramid_config, pyramid_request):
        svc = mock.Mock(spec_set=auth_token_service_factory(None, pyramid_request))
        pyramid_config.register_service(svc, name='auth_token')
        return svc

    @pytest.fixture
    def oauth_token(self, factories):
        authclient = factories.AuthClient(name='Example Client')
        expires = datetime.datetime.utcnow() + datetime.timedelta(minutes=10)
        return factories.DeveloperToken(authclient=authclient, expires=expires)

    @pytest.fixture
    def developer_token(self, factories):
        return factories.DeveloperToken()


class TestAPITokenError(object):
    def test_it_sets_the_response_status_code(self, pyramid_request):
        context = OAuthTokenError('the error message', 'error_type', status_code=403)
        views.api_token_error(context, pyramid_request)
        assert pyramid_request.response.status_code == 403

    def test_it_returns_the_error(self, pyramid_request):
        context = OAuthTokenError('', 'error_type')
        result = views.api_token_error(context, pyramid_request)
        assert result['error'] == 'error_type'

    def test_it_returns_error_description(self, pyramid_request):
        context = OAuthTokenError('error description', 'error_type')
        result = views.api_token_error(context, pyramid_request)
        assert result['error_description'] == 'error description'

    def test_it_skips_description_when_missing(self, pyramid_request):
        context = OAuthTokenError(None, 'invalid_request')
        result = views.api_token_error(context, pyramid_request)
        assert 'error_description' not in result

    def test_it_skips_description_when_empty(self, pyramid_request):
        context = OAuthTokenError('', 'invalid_request')
        result = views.api_token_error(context, pyramid_request)
        assert 'error_description' not in result

[(26, 'pytest.mark.usefixtures', 'pytest.mark.usefixtures', (['"""routes"""', '"""oauth_provider"""', '"""user_svc"""'], {}), False, 'import pytest\n'), (235, 'pytest.mark.usefixtures', 'pytest.mark.usefixtures', (['"""oauth_provider"""'], {}), False, 'import pytest\n'), (273, 'pytest.mark.usefixtures', 'pytest.mark.usefixtures', (['"""oauth_provider"""'], {}), False, 'import pytest\n'), (28, 'pytest.mark.usefixtures', 'pytest.mark.usefixtures', (['"""authenticated_user"""'], {}), False, 'import pytest\n'), (29, 'pytest.mark.parametrize', 'pytest.mark.parametrize', (['"""view_name"""', "['get', 'get_web_message']"], {}), False, 'import pytest\n'), (37, 'pytest.mark.parametrize', 'pytest.mark.parametrize', (['"""view_name"""', "['get', 'get_web_message']"], {}), False, 'import pytest\n'), (47, 'pytest.mark.parametrize', 'pytest.mark.parametrize', (['"""view_name"""', "['get', 'get_web_message']"], {}), False, 'import pytest\n'), (58, 'pytest.mark.parametrize', 'pytest.mark.parametrize', (['"""response_mode,view_name"""', "[(None, 'get'), ('web_message', 'get_web_message')]"], {}), False, 'import pytest\n'), (75, 'pytest.mark.parametrize', 'pytest.mark.parametrize', (['"""view_name"""', "['get', 'get_web_message']"], {}), False, 'import pytest\n'), (95, 'pytest.mark.usefixtures', 'pytest.mark.usefixtures', (['"""authenticated_user"""'], {}), False, 'import pytest\n'), (103, 'pytest.mark.usefixtures', 'pytest.mark.usefixtures', (['"""authenticated_user"""'], {}), False, 'import pytest\n'), (115, 'pytest.mark.usefixtures', 'pytest.mark.usefixtures', (['"""authenticated_user"""'], {}), False, 'import pytest\n'), (125, 'pytest.mark.parametrize', 'pytest.mark.parametrize', (['"""view_name"""', "['post', 'post_web_message']"], {}), False, 'import pytest\n'), (140, 'pytest.mark.usefixtures', 'pytest.mark.usefixtures', (['"""authenticated_user"""'], {}), False, 'import pytest\n'), (141, 'pytest.mark.parametrize', 'pytest.mark.parametrize', (['"""view_name"""', "['post', 'post_web_message']"], {}), False, 'import pytest\n'), (39, 'oauthlib.oauth2.InvalidRequestFatalError', 'InvalidRequestFatalError', (['"""boom!"""'], {}), False, 'from oauthlib.oauth2 import InvalidRequestFatalError\n'), (53, 'h._compat.urlparse.urlparse', 'urlparse.urlparse', (['exc.value.location'], {}), False, 'from h._compat import urlparse\n'), (143, 'oauthlib.oauth2.InvalidRequestFatalError', 'InvalidRequestFatalError', (['"""boom!"""'], {}), False, 'from oauthlib.oauth2 import InvalidRequestFatalError\n'), (178, 'h.views.api_auth.OAuthAuthorizeController', 'views.OAuthAuthorizeController', (['None', 'pyramid_request'], {}), True, 'from h.views import api_auth as views\n'), (182, 'oauthlib.common.Request', 'OAuthRequest', (['"""/"""'], {}), True, 'from oauthlib.common import Request as OAuthRequest\n'), (186, 'mock.create_autospec', 'mock.create_autospec', (['OAuthProviderService'], {'instance': '(True)'}), False, 'import mock\n'), (243, 'json.dumps', 'json.dumps', (["{'access_token': 'the-access-token'}"], {}), False, 'import json\n'), (249, 'json.dumps', 'json.dumps', (["{'error': 'invalid_request'}"], {}), False, 'import json\n'), (263, 'h.views.api_auth.OAuthAccessTokenController', 'views.OAuthAccessTokenController', (['pyramid_request'], {}), True, 'from h.views import api_auth as views\n'), (267, 'mock.Mock', 'mock.Mock', ([], {'spec_set': "['create_token_response']"}), False, 'import mock\n'), (285, 'json.dumps', 'json.dumps', (["{'error': 'invalid_request'}"], {}), False, 'import json\n'), (298, 'h.views.api_auth.OAuthRevocationController', 'views.OAuthRevocationController', (['pyramid_request'], {}), True, 'from h.views import api_auth as views\n'), (302, 'mock.Mock', 'mock.Mock', ([], {'spec_set': "['create_revocation_response']"}), False, 'import mock\n'), (330, 'h.views.api_auth.debug_token', 'views.debug_token', (['pyramid_request'], {}), True, 'from h.views import api_auth as views\n'), (348, 'h.views.api_auth.debug_token', 'views.debug_token', (['pyramid_request'], {}), True, 'from h.views import api_auth as views\n'), (361, 'h.views.api_auth.debug_token', 'views.debug_token', (['pyramid_request'], {}), True, 'from h.views import api_auth as views\n'), (387, 'h.exceptions.OAuthTokenError', 'OAuthTokenError', (['"""the error message"""', '"""error_type"""'], {'status_code': '(403)'}), False, 'from h.exceptions import OAuthTokenError\n'), (388, 'h.views.api_auth.api_token_error', 'views.api_token_error', (['context', 'pyramid_request'], {}), True, 'from h.views import api_auth as views\n'), (392, 'h.exceptions.OAuthTokenError', 'OAuthTokenError', (['""""""', '"""error_type"""'], {}), False, 'from h.exceptions import OAuthTokenError\n'), (393, 'h.views.api_auth.api_token_error', 'views.api_token_error', (['context', 'pyramid_request'], {}), True, 'from h.views import api_auth as views\n'), (397, 'h.exceptions.OAuthTokenError', 'OAuthTokenError', (['"""error description"""', '"""error_type"""'], {}), False, 'from h.exceptions import OAuthTokenError\n'), (398, 'h.views.api_auth.api_token_error', 'views.api_token_error', (['context', 'pyramid_request'], {}), True, 'from h.views import api_auth as views\n'), (402, 'h.exceptions.OAuthTokenError', 'OAuthTokenError', (['None', '"""invalid_request"""'], {}), False, 'from h.exceptions import OAuthTokenError\n'), (403, 'h.views.api_auth.api_token_error', 'views.api_token_error', (['context', 'pyramid_request'], {}), True, 'from h.views import api_auth as views\n'), (407, 'h.exceptions.OAuthTokenError', 'OAuthTokenError', (['""""""', '"""invalid_request"""'], {}), False, 'from h.exceptions import OAuthTokenError\n'), (408, 'h.views.api_auth.api_token_error', 'views.api_token_error', (['context', 'pyramid_request'], {}), True, 'from h.views import api_auth as views\n'), (41, 'pytest.raises', 'pytest.raises', (['InvalidRequestFatalError'], {}), False, 'import pytest\n'), (49, 'pytest.raises', 'pytest.raises', (['httpexceptions.HTTPFound'], {}), False, 'import pytest\n'), (55, 'h._compat.urlparse.parse_qs', 'urlparse.parse_qs', (['parsed_url.query'], {}), False, 'from h._compat import urlparse\n'), (145, 'pytest.raises', 'pytest.raises', (['InvalidRequestFatalError'], {}), False, 'import pytest\n'), (252, 'pytest.raises', 'pytest.raises', (['httpexceptions.HTTPBadRequest'], {}), False, 'import pytest\n'), (288, 'pytest.raises', 'pytest.raises', (['httpexceptions.HTTPBadRequest'], {}), False, 'import pytest\n'), (312, 'pytest.raises', 'pytest.raises', (['OAuthTokenError'], {}), False, 'import pytest\n'), (313, 'h.views.api_auth.debug_token', 'views.debug_token', (['pyramid_request'], {}), True, 'from h.views import api_auth as views\n'), (321, 'pytest.raises', 'pytest.raises', (['OAuthTokenError'], {}), False, 'import pytest\n'), (322, 'h.views.api_auth.debug_token', 'views.debug_token', (['pyramid_request'], {}), True, 'from h.views import api_auth as views\n'), (338, 'pytest.raises', 'pytest.raises', (['OAuthTokenError'], {}), False, 'import pytest\n'), (339, 'h.views.api_auth.debug_token', 'views.debug_token', (['pyramid_request'], {}), True, 'from h.views import api_auth as views\n'), (377, 'datetime.datetime.utcnow', 'datetime.datetime.utcnow', ([], {}), False, 'import datetime\n'), (377, 'datetime.timedelta', 'datetime.timedelta', ([], {'minutes': '(10)'}), False, 'import datetime\n'), (209, 'h.services.user.user_service_factory', 'user_service_factory', (['None', 'pyramid_request'], {}), False, 'from h.services.user import user_service_factory\n'), (353, 'h.util.datetime.utc_iso8601', 'utc_iso8601', (['oauth_token.created'], {}), False, 'from h.util.datetime import utc_iso8601\n'), (354, 'h.util.datetime.utc_iso8601', 'utc_iso8601', (['oauth_token.expires'], {}), False, 'from h.util.datetime import utc_iso8601\n'), (364, 'h.util.datetime.utc_iso8601', 'utc_iso8601', (['developer_token.created'], {}), False, 'from h.util.datetime import utc_iso8601\n'), (370, 'h.services.auth_token.auth_token_service_factory', 'auth_token_service_factory', (['None', 'pyramid_request'], {}), False, 'from h.services.auth_token import auth_token_service_factory\n')]