# coding: utf-8
import asyncio
import json

from asynctest import CoroutineMock
from sanic import Sanic
from pytest_sanic.utils import TestClient as SanicTestClient
from sanic_plugin_toolkit import SanicPluginRealm

from .server import (
    create_server,
    db,
    cache_provider,
    bind_cache_provider,
    sqlalchemy_provider,
    default_provider,
    Token
)
from .client import create_client
from .._base import BaseSuite, clean_url, to_base64
from .._base import to_unicode as u


class OAuthSuite(BaseSuite):
    @property
    def database(self):
        return db

    def create_oauth_provider(self, app):
        raise NotImplementedError('Each test class must'
                                  'implement this method.')

    async def bind_oauth_with_server(self, oauth, app):
        pass

    def create_app(self):
        client_app = Sanic("{}_client".format(self.id()))
        server_app = Sanic("{}_server".format(self.id()))
        realm1 = SanicPluginRealm(client_app)
        realm2 = SanicPluginRealm(server_app)
        return server_app, client_app

    async def setup_app(self, server_app, client_app):
        oauth = self.create_oauth_provider(server_app)
        loop = asyncio.get_event_loop_policy().get_event_loop()
        self.server_client = SanicTestClient(server_app, loop)
        create_server(server_app, oauth)
        self.oauth_client = create_client(client_app)
        self.oauth_client.http_request = CoroutineMock(
            side_effect=self.patch_request(self.server_client)
        )
        await self.server_client.start_server()
        await self.bind_oauth_with_server(oauth, server_app)
        return server_app

    async def tearDown(self):
        await self.server_client.close()
        await super(OAuthSuite, self).tearDown()


authorize_url = (
    '/oauth2/authorize?response_type=code&client_id=dev'
    '&redirect_uri=http%3A%2F%2F127.0.0.1%3A5000%2Fauthorized&scope=email'
)


auth_code = to_base64('confidential:confidential')


class TestWebAuth(OAuthSuite):

    def create_oauth_provider(self, app):
        return default_provider(app)

    async def test_login(self):
        rv = await self.sanic_client.get('/login', allow_redirects=False)
        location = rv.headers.get('Location')
        assert 'response_type=code' in location

    async def test_oauth_authorize_invalid_url(self):
        resp, content = await self.oauth_client.http_request('/oauth2/authorize', allow_redirects=False)
        location = resp.headers.get('Location')
        assert 'Missing+client_id+parameter.' in location

    async def test_oauth_authorize_valid_url(self):
        resp, content = await self.oauth_client.http_request(authorize_url)
        assert '</form>' in content

        resp, content = await self.oauth_client.http_request(authorize_url, data=dict(
            confirm='no'
        ), method='post', allow_redirects=False)
        location = resp.headers.get('Location')
        assert 'access_denied' in location

        resp, content = await self.oauth_client.http_request(authorize_url, data=dict(
            confirm='yes'
        ), method='post', allow_redirects=False)
        # success
        location = resp.headers.get('Location')
        assert 'code=' in location
        assert 'state' not in location

        # test state on access denied
        # According to RFC 6749, state should be preserved on error response if it's present in the client request.
        # Reference: https://tools.ietf.org/html/rfc6749#section-4.1.2
        resp, content = await self.oauth_client.http_request(authorize_url + '&state=foo', data=dict(
            confirm='no'
        ), method='post', allow_redirects=False)
        location = resp.headers.get('Location')
        assert 'error=access_denied' in location
        assert 'state=foo' in location

        # test state on success
        resp, content = await self.oauth_client.http_request(authorize_url + '&state=foo', data=dict(
            confirm='yes'
        ), method='post', allow_redirects=False)
        location = resp.headers.get('Location')
        assert 'code=' in location
        assert 'state=foo' in location

    async def test_http_head_oauth_authorize_valid_url(self):
        resp, content = await self.oauth_client.http_request(authorize_url, method='head', allow_redirects=False)
        assert resp.headers['X-Client-ID'] == 'dev'

    async def test_get_access_token(self):
        resp, content = await self.oauth_client.http_request(authorize_url,
            data={'confirm': 'yes'}, method='post', allow_redirects=False)
        location = resp.headers.get('Location')
        rv = await self.sanic_client.get(clean_url(location))
        assert 'access_token' in rv.text

    async def test_full_flow(self):
        resp, content = await self.oauth_client.http_request(authorize_url,
            data={'confirm': 'yes'}, method='post', allow_redirects=False)
        location = resp.headers.get('Location')
        rv = await self.sanic_client.get(clean_url(location))
        content = rv.text
        assert 'access_token' in content

        rv = await self.sanic_client.get('/')
        content = rv.text
        assert 'username' in content

        rv = await self.sanic_client.get('/address')
        assert rv.status_code == 401
        content = rv.text
        assert 'message' in content

        rv = await self.sanic_client.get('/method/post')
        content = rv.text
        assert 'POST' in content

        rv = await self.sanic_client.get('/method/put')
        content = rv.text
        assert 'PUT' in content

        rv = await self.sanic_client.get('/method/delete')
        content = rv.text
        assert 'DELETE' in content

    async def test_no_bear_token(self):
        @self.oauth_client.tokengetter
        def get_oauth_token():
            return 'foo', ''

        rv = await self.sanic_client.get('/method/put')
        content = rv.text
        assert 'token not found' in content

    async def test_expires_bear_token(self):
        @self.oauth_client.tokengetter
        def get_oauth_token():
            return 'expired', ''

        rv = await self.sanic_client.get('/method/put')
        assert 'token is expired' in rv.text

    async def test_never_expiring_bear_token(self):
        @self.oauth_client.tokengetter
        def get_oauth_token():
            return 'never_expire', ''

        rv = await self.sanic_client.get('/method/put')
        assert rv.status_code == 200

    async def test_get_client(self):
        resp, content = await self.oauth_client.http_request(authorize_url,
            data={'confirm': 'yes'}, method='post', allow_redirects=False)
        location = resp.headers.get('Location')
        rv = await self.sanic_client.get(clean_url(location))
        content1 = rv.text
        rv = await self.sanic_client.get("/client")
        content2 = rv.text
        assert 'dev' in content2

    async def test_invalid_response_type(self):
        authorize_url = (
            '/oauth2/authorize?response_type=invalid&client_id=dev'
            '&redirect_uri=http%3A%2F%2F127.0.0.1%3A5000%2Fauthorized'
            '&scope=email'
        )
        resp, content = await self.oauth_client.http_request(authorize_url,
            data={'confirm': 'yes'}, method='post', allow_redirects=False)
        location = resp.headers.get('Location')
        rv = await self.sanic_client.get(clean_url(location))
        assert 'error' in rv.text

    async def test_invalid_scope(self):
        authorize_url = (
            '/oauth2/authorize?response_type=code&client_id=dev'
            '&redirect_uri=http%3A%2F%2F127.0.0.1%3A5000%2Fauthorized'
            '&scope=invalid'
        )
        resp, content = await self.oauth_client.http_request(authorize_url,
            data={'confirm': 'yes'}, method='post', allow_redirects=False)
        location = resp.headers.get('Location')
        rv = await self.sanic_client.get(clean_url(location))
        assert 'error' in rv.text
        assert 'invalid_scope' in rv.text


class TestWebAuthCached(TestWebAuth):

    def create_oauth_provider(self, app):
        return cache_provider(app)

    async def bind_oauth_with_server(self, oauth, app):
        return bind_cache_provider(oauth, app)


class TestWebAuthSQLAlchemy(TestWebAuth):

    def create_oauth_provider(self, app):
        return sqlalchemy_provider(app)


class TestRefreshToken(OAuthSuite):

    def create_oauth_provider(self, app):
        return default_provider(app)

    async def test_refresh_token_in_password_grant(self):
        resp, content = await self.oauth_client.http_request("/oauth2/token",
            data={"grant_type": "password", "scope": "email address", "username": "admin", "password": "admin"},
            headers={'Authorization': 'Basic %s' % auth_code,}, method="POST")
        assert 'access_token' in resp.text
        data = json.loads(resp.text)

        resp, content = await self.oauth_client.http_request('/oauth2/token', data={
            "grant_type": "refresh_token",
            "scope": data.get('scope'),
            "refresh_token":  data.get('refresh_token'),
        }, headers={'Authorization': 'Basic %s' % auth_code,})
        assert 'access_token' in resp.text


    async def test_refresh_token_in_authorization_code(self):
        resp, content = await self.oauth_client.http_request(authorize_url,
                                                             data={'confirm': 'yes'}, method='post',
                                                             allow_redirects=False)
        location = resp.headers.get('Location')
        rv = await self.sanic_client.get(clean_url(location))
        data = json.loads(rv.text)

        resp, content = await self.oauth_client.http_request('/oauth2/token', data={
            "grant_type": "refresh_token",
            "scope": data.get('scope'),
            "refresh_token":  data.get('refresh_token'),
            "client_id": "dev", "client_secret": "dev"
        })
        assert b'access_token' in resp._body


class TestRefreshTokenCached(TestRefreshToken):

    def create_oauth_provider(self, app):
        return cache_provider(app)


class TestRefreshTokenSQLAlchemy(TestRefreshToken):

    def create_oauth_provider(self, app):
        return sqlalchemy_provider(app)


class TestRevokeToken(OAuthSuite):

    def create_oauth_provider(self, app):
        return default_provider(app)

    def get_token(self):
        url = ('/oauth/token?grant_type=password'
               '&scope=email+address&username=admin&password=admin')
        rv = self.sanic_client.get(url, headers={
            'Authorization': 'Basic %s' % auth_code,
        })
        assert b'_token' in rv.data
        return json.loads(u(rv.data))

    def test_revoke_token(self):
        data = self.get_token()
        tok = Token.query.filter_by(
            refresh_token=data['refresh_token']).first()
        assert tok.refresh_token == data['refresh_token']

        revoke_url = '/oauth/revoke'
        args = {'token': data['refresh_token']}
        self.sanic_client.post(revoke_url, data=args, headers={
            'Authorization': 'Basic %s' % auth_code,
        })

        tok = Token.query.filter_by(
            refresh_token=data['refresh_token']).first()
        assert tok is None

    def test_revoke_token_with_hint(self):
        data = self.get_token()
        tok = Token.query.filter_by(
            access_token=data['access_token']).first()
        assert tok.access_token == data['access_token']

        revoke_url = '/oauth/revoke'
        args = {'token': data['access_token'],
                'token_type_hint': 'access_token'}
        self.sanic_client.post(revoke_url, data=args, headers={
            'Authorization': 'Basic %s' % auth_code,
        })

        tok = Token.query.filter_by(
            access_token=data['access_token']).first()
        assert tok is None


class TestRevokeTokenCached(TestRefreshToken):

    def create_oauth_provider(self, app):
        return cache_provider(app)


class TestRevokeTokenSQLAlchemy(TestRefreshToken):

    def create_oauth_provider(self, app):
        return sqlalchemy_provider(app)


class TestCredentialAuth(OAuthSuite):

    def create_oauth_provider(self, app):
        return default_provider(app)

    def test_get_access_token(self):
        url = ('/oauth/token?grant_type=client_credentials'
               '&scope=email+address')
        (rq, rv) = self.sanic_client.get(url, headers={
            'Authorization': 'Basic %s' % auth_code,
        })
        assert b'access_token' in rv.data

    def test_invalid_auth_header(self):
        url = ('/oauth/token?grant_type=client_credentials'
               '&scope=email+address')
        (rq, rv) = self.sanic_client.get(url, headers={
            'Authorization': 'Basic foobar'
        })
        assert b'invalid_client' in rv.data

    def test_no_client(self):
        auth_code = to_base64('none:confidential')
        url = ('/oauth/token?grant_type=client_credentials'
               '&scope=email+address')
        (rq, rv) = self.sanic_client.get(url, headers={
            'Authorization': 'Basic %s' % auth_code,
        })
        assert b'invalid_client' in rv.data

    def test_wrong_secret_client(self):
        auth_code = to_base64('confidential:wrong')
        url = ('/oauth/token?grant_type=client_credentials'
               '&scope=email+address')
        (rq, rv) = self.sanic_client.get(url, headers={
            'Authorization': 'Basic %s' % auth_code,
        })
        assert b'invalid_client' in rv.data


class TestCredentialAuthCached(TestCredentialAuth):

    def create_oauth_provider(self, app):
        return cache_provider(app)


class TestCredentialAuthSQLAlchemy(TestCredentialAuth):

    def create_oauth_provider(self, app):
        return sqlalchemy_provider(app)


class TestTokenGenerator(OAuthSuite):

    def create_oauth_provider(self, app):

        def generator(request):
            return 'foobar'

        app.config['OAUTH2_PROVIDER_TOKEN_GENERATOR'] = generator
        return default_provider(app)

    def test_get_access_token(self):
        (rq, rv) = self.sanic_client.post(authorize_url, data={'confirm': 'yes'})
        (rq, rv) = self.sanic_client.get(clean_url(location))
        data = json.loads(u(rv.data))
        assert data['access_token'] == 'foobar'
        assert data['refresh_token'] == 'foobar'


class TestRefreshTokenGenerator(OAuthSuite):

    def create_oauth_provider(self, app):

        def at_generator(request):
            return 'foobar'

        def rt_generator(request):
            return 'abracadabra'

        app.config['OAUTH2_PROVIDER_TOKEN_GENERATOR'] = at_generator
        app.config['OAUTH2_PROVIDER_REFRESH_TOKEN_GENERATOR'] = rt_generator
        return default_provider(app)

    def test_get_access_token(self):
        (rq, rv) = self.sanic_client.post(authorize_url, data={'confirm': 'yes'})
        (rq, rv) = self.sanic_client.get(clean_url(location))
        data = json.loads(u(rv.data))
        assert data['access_token'] == 'foobar'
        assert data['refresh_token'] == 'abracadabra'


class TestConfidentialClient(OAuthSuite):

    def create_oauth_provider(self, app):
        return default_provider(app)

    async def test_get_access_token(self):
        url = ('/oauth/token?grant_type=authorization_code&code=12345'
               '&scope=email')
        rv = await self.sanic_client.get(url, headers={
            'Authorization': 'Basic %s' % auth_code
        })
        await rv.read()
        assert b'access_token' in rv.data

    async def test_invalid_grant(self):
        url = ('/oauth/token?grant_type=authorization_code&code=54321'
               '&scope=email')
        rv = await self.sanic_client.get(url, headers={
            'Authorization': 'Basic %s' % auth_code
        })
        await rv.read()
        assert b'invalid_grant' in rv.data

    async def test_invalid_client(self):
        url = ('/oauth/token?grant_type=authorization_code&code=12345'
               '&scope=email')
        rv = await self.sanic_client.get(url, headers={
            'Authorization': 'Basic %s' % ('foo')
        })
        await rv.read()
        assert b'invalid_client' in rv.data
