import io
from contextlib import redirect_stdout
from unittest.mock import patch

import pytest

from peony import oauth, oauth_dance, utils

alphabet = "".join(chr(i) for i in range(ord("a"), ord("a") + 26))


class MockOAuth2Headers(oauth.OAuth2Headers):
    def __init__(self, consumer_key, consumer_secret, client=None, **kwargs):
        assert consumer_key == "a"
        assert consumer_secret == "b"
        super().__init__(consumer_key, consumer_secret, client)

    async def sign(self, *args, **kwargs):
        self.token = alphabet
        return self.copy()


def test_oauth2_dance(event_loop):
    with patch.object(
        oauth_dance.oauth, "OAuth2Headers", side_effect=MockOAuth2Headers
    ):
        args = "consumer_key", "consumer_secret"
        with patch.object(utils, "get_args", return_value=args):
            assert alphabet == oauth_dance.oauth2_dance("a", "b", event_loop)


@pytest.mark.asyncio
async def test_get_oauth_token():
    async def request(method, url, future, *args, **kwargs):
        assert method == "post"
        assert url == "https://api.twitter.com/oauth/request_token"
        data = "oauth_token=abc&hello=world"
        future.set_result(data)
        return data

    with patch.object(oauth_dance.BasePeonyClient, "request", side_effect=request):
        data = await oauth_dance.get_oauth_token("", "")
        assert data == {"oauth_token": "abc", "hello": "world"}


async def dummy(*args, **kwargs):
    pass


@pytest.mark.asyncio
async def test_get_oauth_verifier():
    def test_url(url):
        assert url == "https://api.twitter.com/oauth/authorize?oauth_token=abc"
        return True

    with patch.object(oauth_dance.webbrowser, "open", side_effect=test_url):
        with patch.object(oauth_dance.asyncio, "sleep", side_effect=dummy):
            with patch.object(oauth_dance, "input", return_value="12345"):
                assert await oauth_dance.get_oauth_verifier("abc") == "12345"


@pytest.mark.asyncio
async def test_get_oauth_verifier_no_browser():
    stream = io.StringIO()
    url = "https://api.twitter.com/oauth/authorize?oauth_token=abc"

    with patch.object(oauth_dance.asyncio, "sleep", side_effect=dummy):
        with patch.object(oauth_dance.webbrowser, "open", return_value=False):
            with patch.object(oauth_dance, "input", return_value="12345"):
                with redirect_stdout(stream):
                    await oauth_dance.get_oauth_verifier("abc")

                stream.seek(0)
                assert url in stream.read()


@pytest.mark.asyncio
async def test_get_oauth_verifier_browser_error():
    def error(url):
        raise RuntimeError

    stream = io.StringIO()
    url = "https://api.twitter.com/oauth/authorize?oauth_token=abc"

    with patch.object(oauth_dance.asyncio, "sleep", side_effect=dummy):
        with patch.object(oauth_dance.webbrowser, "open", side_effect=error):
            with patch.object(oauth_dance, "input", return_value="12345"):
                with redirect_stdout(stream):
                    await oauth_dance.get_oauth_verifier("abc")

                stream.seek(0)
                assert url in stream.read()


@pytest.mark.asyncio
async def test_get_access_token():
    async def request(method, url, params, future, *args, **kwargs):
        assert method == "get"
        assert url == "https://api.twitter.com/oauth/access_token"
        assert params["oauth_verifier"] == "12345"
        data = "access_token=abc&access_secret=cba"
        future.set_result(data)
        return data

    with patch.object(oauth_dance.BasePeonyClient, "request", side_effect=request):
        data = await oauth_dance.get_access_token("", "", "", "", "12345")
        assert data == {"access_token": "abc", "access_secret": "cba"}


@pytest.mark.asyncio
async def test_async_oauth_dance():
    with patch.object(oauth_dance, "get_oauth_token") as get_oauth_token:
        with patch.object(oauth_dance, "get_oauth_verifier") as get_oauth_verifier:
            with patch.object(oauth_dance, "get_access_token") as get_access_token:

                async def oauth_token(consumer_key, consumer_secret, callback_uri):
                    assert consumer_key == "a"
                    assert consumer_secret == "b"
                    assert callback_uri == "oob"

                    assert not get_oauth_verifier.called
                    assert not get_access_token.called

                    return {"oauth_token": alphabet, "another": "cba"}

                async def oauth_verifier(oauth_token):
                    assert oauth_token == alphabet

                    assert get_oauth_token.called
                    assert not get_access_token.called
                    return "12345"

                async def access_token(
                    consumer_key, consumer_secret, oauth_verifier, oauth_token, another
                ):
                    assert consumer_key == "a"
                    assert consumer_secret == "b"
                    assert oauth_verifier == "12345"
                    assert oauth_token == alphabet
                    assert another == "cba"

                    assert get_access_token.called
                    assert get_oauth_verifier.called

                    return {
                        "consumer_key": consumer_key,
                        "consumer_secret": consumer_secret,
                        "oauth_token": "abcdef",
                        "oauth_token_secret": "ghijkl",
                        "hello": "world",
                    }

                get_oauth_verifier.side_effect = oauth_verifier
                get_oauth_token.side_effect = oauth_token
                get_access_token.side_effect = access_token

                tokens = await oauth_dance.async_oauth_dance("a", "b")

                assert tokens == {
                    "consumer_key": "a",
                    "consumer_secret": "b",
                    "access_token": "abcdef",
                    "access_token_secret": "ghijkl",
                }


def test_oauth_dance(event_loop):
    data = {
        "consumer_key": "abc",
        "consumer_secret": "def",
        "access_token": "ghi",
        "access_token_secret": "jkl",
    }

    async def async_oauth_dance(consumer_key, consumer_secret, callback_uri):
        return data

    with patch.object(
        oauth_dance, "async_oauth_dance", side_effect=async_oauth_dance
    ) as async_dance:
        assert oauth_dance.oauth_dance("a", "b", loop=event_loop) == data
        assert async_dance.called_with("a", "b", "oob")
