import unittest
import urllib.error

from websockets.auth import *
from websockets.auth import is_credentials
from websockets.exceptions import InvalidStatusCode
from websockets.headers import build_authorization_basic

from .test_client_server import ClientServerTestsMixin, with_client, with_server
from .utils import AsyncioTestCase


class AuthTests(unittest.TestCase):
    def test_is_credentials(self):
        self.assertTrue(is_credentials(("username", "password")))

    def test_is_not_credentials(self):
        self.assertFalse(is_credentials(None))
        self.assertFalse(is_credentials("username"))


class AuthClientServerTests(ClientServerTestsMixin, AsyncioTestCase):

    create_protocol = basic_auth_protocol_factory(
        realm="auth-tests", credentials=("hello", "iloveyou")
    )

    @with_server(create_protocol=create_protocol)
    @with_client(user_info=("hello", "iloveyou"))
    def test_basic_auth(self):
        req_headers = self.client.request_headers
        resp_headers = self.client.response_headers
        self.assertEqual(req_headers["Authorization"], "Basic aGVsbG86aWxvdmV5b3U=")
        self.assertNotIn("WWW-Authenticate", resp_headers)

        self.loop.run_until_complete(self.client.send("Hello!"))
        self.loop.run_until_complete(self.client.recv())

    def test_basic_auth_server_no_credentials(self):
        with self.assertRaises(TypeError) as raised:
            basic_auth_protocol_factory(realm="auth-tests", credentials=None)
        self.assertEqual(
            str(raised.exception), "provide either credentials or check_credentials"
        )

    def test_basic_auth_server_bad_credentials(self):
        with self.assertRaises(TypeError) as raised:
            basic_auth_protocol_factory(realm="auth-tests", credentials=42)
        self.assertEqual(str(raised.exception), "invalid credentials argument: 42")

    create_protocol_multiple_credentials = basic_auth_protocol_factory(
        realm="auth-tests",
        credentials=[("hello", "iloveyou"), ("goodbye", "stillloveu")],
    )

    @with_server(create_protocol=create_protocol_multiple_credentials)
    @with_client(user_info=("hello", "iloveyou"))
    def test_basic_auth_server_multiple_credentials(self):
        self.loop.run_until_complete(self.client.send("Hello!"))
        self.loop.run_until_complete(self.client.recv())

    def test_basic_auth_bad_multiple_credentials(self):
        with self.assertRaises(TypeError) as raised:
            basic_auth_protocol_factory(
                realm="auth-tests", credentials=[("hello", "iloveyou"), 42]
            )
        self.assertEqual(
            str(raised.exception),
            "invalid credentials argument: [('hello', 'iloveyou'), 42]",
        )

    async def check_credentials(username, password):
        return password == "iloveyou"

    create_protocol_check_credentials = basic_auth_protocol_factory(
        realm="auth-tests", check_credentials=check_credentials
    )

    @with_server(create_protocol=create_protocol_check_credentials)
    @with_client(user_info=("hello", "iloveyou"))
    def test_basic_auth_check_credentials(self):
        self.loop.run_until_complete(self.client.send("Hello!"))
        self.loop.run_until_complete(self.client.recv())

    @with_server(create_protocol=create_protocol)
    def test_basic_auth_missing_credentials(self):
        with self.assertRaises(InvalidStatusCode) as raised:
            self.start_client()
        self.assertEqual(raised.exception.status_code, 401)

    @with_server(create_protocol=create_protocol)
    def test_basic_auth_missing_credentials_details(self):
        with self.assertRaises(urllib.error.HTTPError) as raised:
            self.loop.run_until_complete(self.make_http_request())
        self.assertEqual(raised.exception.code, 401)
        self.assertEqual(
            raised.exception.headers["WWW-Authenticate"],
            'Basic realm="auth-tests", charset="UTF-8"',
        )
        self.assertEqual(raised.exception.read().decode(), "Missing credentials\n")

    @with_server(create_protocol=create_protocol)
    def test_basic_auth_unsupported_credentials(self):
        with self.assertRaises(InvalidStatusCode) as raised:
            self.start_client(extra_headers={"Authorization": "Digest ..."})
        self.assertEqual(raised.exception.status_code, 401)

    @with_server(create_protocol=create_protocol)
    def test_basic_auth_unsupported_credentials_details(self):
        with self.assertRaises(urllib.error.HTTPError) as raised:
            self.loop.run_until_complete(
                self.make_http_request(headers={"Authorization": "Digest ..."})
            )
        self.assertEqual(raised.exception.code, 401)
        self.assertEqual(
            raised.exception.headers["WWW-Authenticate"],
            'Basic realm="auth-tests", charset="UTF-8"',
        )
        self.assertEqual(raised.exception.read().decode(), "Unsupported credentials\n")

    @with_server(create_protocol=create_protocol)
    def test_basic_auth_invalid_credentials(self):
        with self.assertRaises(InvalidStatusCode) as raised:
            self.start_client(user_info=("hello", "ihateyou"))
        self.assertEqual(raised.exception.status_code, 401)

    @with_server(create_protocol=create_protocol)
    def test_basic_auth_invalid_credentials_details(self):
        with self.assertRaises(urllib.error.HTTPError) as raised:
            authorization = build_authorization_basic("hello", "ihateyou")
            self.loop.run_until_complete(
                self.make_http_request(headers={"Authorization": authorization})
            )
        self.assertEqual(raised.exception.code, 401)
        self.assertEqual(
            raised.exception.headers["WWW-Authenticate"],
            'Basic realm="auth-tests", charset="UTF-8"',
        )
        self.assertEqual(raised.exception.read().decode(), "Invalid credentials\n")
