import asyncio
import json
import re
from time import time
from urllib.parse import quote

import pytest
from slack_sdk.oauth.installation_store import FileInstallationStore
from slack_sdk.oauth.state_store import FileOAuthStateStore
from slack_sdk.signature import SignatureVerifier
from slack_sdk.web.async_client import AsyncWebClient

from slack_bolt import BoltResponse
from slack_bolt.app.async_app import AsyncApp
from slack_bolt.oauth.async_callback_options import (
    AsyncFailureArgs,
    AsyncSuccessArgs,
    AsyncCallbackOptions,
)
from slack_bolt.oauth.async_oauth_flow import AsyncOAuthFlow
from slack_bolt.oauth.async_oauth_settings import AsyncOAuthSettings
from slack_bolt.request.async_request import AsyncBoltRequest
from tests.mock_web_api_server import (
    cleanup_mock_web_api_server,
    setup_mock_web_api_server,
)


class TestAsyncOAuthFlow:
    mock_api_server_base_url = "http://localhost:8888"

    @pytest.fixture
    def event_loop(self):
        setup_mock_web_api_server(self)
        loop = asyncio.get_event_loop()
        yield loop
        loop.close()
        cleanup_mock_web_api_server(self)

    def next(self):
        pass

    @pytest.mark.asyncio
    async def test_instantiation(self):
        oauth_flow = AsyncOAuthFlow(
            settings=AsyncOAuthSettings(
                client_id="111.222",
                client_secret="xxx",
                scopes=["chat:write", "commands"],
                installation_store=FileInstallationStore(),
                state_store=FileOAuthStateStore(expiration_seconds=120),
            )
        )
        assert oauth_flow is not None
        assert oauth_flow.logger is not None
        assert oauth_flow.client is not None

    @pytest.mark.asyncio
    async def test_handle_installation(self):
        oauth_flow = AsyncOAuthFlow(
            settings=AsyncOAuthSettings(
                client_id="111.222",
                client_secret="xxx",
                scopes=["chat:write", "commands"],
                installation_store=FileInstallationStore(),
                state_store=FileOAuthStateStore(expiration_seconds=120),
            )
        )
        req = AsyncBoltRequest(body="")
        resp = await oauth_flow.handle_installation(req)
        assert resp.status == 302
        url = resp.headers["location"][0]
        assert (
            re.compile(
                "https://slack.com/oauth/v2/authorize\\?state=[-0-9a-z]+."
                "&client_id=111\\.222"
                "&scope=chat:write,commands"
                "&user_scope="
            ).match(url)
            is not None
        )

    @pytest.mark.asyncio
    async def test_handle_callback(self):
        oauth_flow = AsyncOAuthFlow(
            client=AsyncWebClient(base_url=self.mock_api_server_base_url),
            settings=AsyncOAuthSettings(
                client_id="111.222",
                client_secret="xxx",
                scopes=["chat:write", "commands"],
                installation_store=FileInstallationStore(),
                state_store=FileOAuthStateStore(expiration_seconds=120),
                success_url="https://www.example.com/completion",
                failure_url="https://www.example.com/failure",
            ),
        )
        state = await oauth_flow.issue_new_state(None)
        req = AsyncBoltRequest(
            body="",
            query=f"code=foo&state={state}",
            headers={"cookie": [f"{oauth_flow.settings.state_cookie_name}={state}"]},
        )
        resp = await oauth_flow.handle_callback(req)
        assert resp.status == 200
        assert "https://www.example.com/completion" in resp.body

        app = AsyncApp(signing_secret="signing_secret", oauth_flow=oauth_flow)
        global_shortcut_body = {
            "type": "shortcut",
            "token": "verification_token",
            "action_ts": "111.111",
            "team": {
                "id": "T111",
                "domain": "workspace-domain",
                "enterprise_id": "E111",
                "enterprise_name": "Org Name",
            },
            "user": {"id": "W111", "username": "primary-owner", "team_id": "T111"},
            "callback_id": "test-shortcut",
            "trigger_id": "111.111.xxxxxx",
        }
        body = f"payload={quote(json.dumps(global_shortcut_body))}"
        timestamp = str(int(time()))
        signature_verifier = SignatureVerifier("signing_secret")
        headers = {
            "content-type": ["application/x-www-form-urlencoded"],
            "x-slack-signature": [
                signature_verifier.generate_signature(body=body, timestamp=timestamp)
            ],
            "x-slack-request-timestamp": [timestamp],
        }
        request = AsyncBoltRequest(body=body, headers=headers)
        response = await app.async_dispatch(request)
        assert response.status == 200
        assert self.mock_received_requests["/auth.test"] == 1

    @pytest.mark.asyncio
    async def test_handle_callback_invalid_state(self):
        oauth_flow = AsyncOAuthFlow(
            settings=AsyncOAuthSettings(
                client_id="111.222",
                client_secret="xxx",
                scopes=["chat:write", "commands"],
                installation_store=FileInstallationStore(),
                state_store=FileOAuthStateStore(expiration_seconds=120),
            )
        )
        state = await oauth_flow.issue_new_state(None)
        req = AsyncBoltRequest(
            body="",
            query=f"code=foo&state=invalid",
            headers={"cookie": [f"{oauth_flow.settings.state_cookie_name}={state}"]},
        )
        resp = await oauth_flow.handle_callback(req)
        assert resp.status == 400

    @pytest.mark.asyncio
    async def test_handle_callback_using_options(self):
        async def success(args: AsyncSuccessArgs) -> BoltResponse:
            assert args.request is not None
            return BoltResponse(status=200, body="customized")

        async def failure(args: AsyncFailureArgs) -> BoltResponse:
            assert args.request is not None
            assert args.reason is not None
            return BoltResponse(status=502, body="customized")

        oauth_flow = AsyncOAuthFlow(
            client=AsyncWebClient(base_url=self.mock_api_server_base_url),
            settings=AsyncOAuthSettings(
                client_id="111.222",
                client_secret="xxx",
                scopes=["chat:write", "commands"],
                installation_store=FileInstallationStore(),
                state_store=FileOAuthStateStore(expiration_seconds=120),
                callback_options=AsyncCallbackOptions(
                    success=success, failure=failure,
                ),
            ),
        )
        state = await oauth_flow.issue_new_state(None)
        req = AsyncBoltRequest(
            body="",
            query=f"code=foo&state={state}",
            headers={"cookie": [f"{oauth_flow.settings.state_cookie_name}={state}"]},
        )
        resp = await oauth_flow.handle_callback(req)
        assert resp.status == 200
        assert resp.body == "customized"

        req = AsyncBoltRequest(
            body="",
            query=f"code=foo&state=invalid",
            headers={"cookie": [f"{oauth_flow.settings.state_cookie_name}={state}"]},
        )
        resp = await oauth_flow.handle_callback(req)
        assert resp.status == 502
        assert resp.body == "customized"
