from __future__ import annotations

import base64
import re
import uuid

from dataclasses import dataclass
from pathlib import Path
from typing import TYPE_CHECKING
from typing import Any

import httpretty
import pytest
import requests

from cleo.io.null_io import NullIO

from poetry.utils.authenticator import Authenticator


if TYPE_CHECKING:
    from _pytest.monkeypatch import MonkeyPatch
    from pytest_mock import MockerFixture

    from tests.conftest import Config
    from tests.conftest import DummyBackend


@dataclass
class SimpleCredential:
    username: str
    password: str


@pytest.fixture()
def mock_remote(http: type[httpretty.httpretty]) -> None:
    http.register_uri(
        http.GET,
        re.compile("^https?://foo.bar/(.+?)$"),
    )


def test_authenticator_uses_url_provided_credentials(
    config: Config, mock_remote: None, http: type[httpretty.httpretty]
):
    config.merge(
        {
            "repositories": {"foo": {"url": "https://foo.bar/simple/"}},
            "http-basic": {"foo": {"username": "bar", "password": "baz"}},
        }
    )

    authenticator = Authenticator(config, NullIO())
    authenticator.request("get", "https://foo001:bar002@foo.bar/files/foo-0.1.0.tar.gz")

    request = http.last_request()

    assert request.headers["Authorization"] == "Basic Zm9vMDAxOmJhcjAwMg=="


def test_authenticator_uses_credentials_from_config_if_not_provided(
    config: Config, mock_remote: None, http: type[httpretty.httpretty]
):
    config.merge(
        {
            "repositories": {"foo": {"url": "https://foo.bar/simple/"}},
            "http-basic": {"foo": {"username": "bar", "password": "baz"}},
        }
    )

    authenticator = Authenticator(config, NullIO())
    authenticator.request("get", "https://foo.bar/files/foo-0.1.0.tar.gz")

    request = http.last_request()

    assert request.headers["Authorization"] == "Basic YmFyOmJheg=="


def test_authenticator_uses_username_only_credentials(
    config: Config,
    mock_remote: None,
    http: type[httpretty.httpretty],
    with_simple_keyring: None,
):
    config.merge(
        {
            "repositories": {"foo": {"url": "https://foo.bar/simple/"}},
            "http-basic": {"foo": {"username": "bar", "password": "baz"}},
        }
    )

    authenticator = Authenticator(config, NullIO())
    authenticator.request("get", "https://foo001@foo.bar/files/foo-0.1.0.tar.gz")

    request = http.last_request()

    assert request.headers["Authorization"] == "Basic Zm9vMDAxOg=="


def test_authenticator_uses_password_only_credentials(
    config: Config, mock_remote: None, http: type[httpretty.httpretty]
):
    config.merge(
        {
            "repositories": {"foo": {"url": "https://foo.bar/simple/"}},
            "http-basic": {"foo": {"username": "bar", "password": "baz"}},
        }
    )

    authenticator = Authenticator(config, NullIO())
    authenticator.request("get", "https://:bar002@foo.bar/files/foo-0.1.0.tar.gz")

    request = http.last_request()

    assert request.headers["Authorization"] == "Basic OmJhcjAwMg=="


def test_authenticator_uses_empty_strings_as_default_password(
    config: Config,
    mock_remote: None,
    http: type[httpretty.httpretty],
    with_simple_keyring: None,
):
    config.merge(
        {
            "repositories": {"foo": {"url": "https://foo.bar/simple/"}},
            "http-basic": {"foo": {"username": "bar"}},
        }
    )

    authenticator = Authenticator(config, NullIO())
    authenticator.request("get", "https://foo.bar/files/foo-0.1.0.tar.gz")

    request = http.last_request()

    assert request.headers["Authorization"] == "Basic YmFyOg=="


def test_authenticator_uses_empty_strings_as_default_username(
    config: Config, mock_remote: None, http: type[httpretty.httpretty]
):
    config.merge(
        {
            "repositories": {"foo": {"url": "https://foo.bar/simple/"}},
            "http-basic": {"foo": {"username": None, "password": "bar"}},
        }
    )

    authenticator = Authenticator(config, NullIO())
    authenticator.request("get", "https://foo.bar/files/foo-0.1.0.tar.gz")

    request = http.last_request()

    assert request.headers["Authorization"] == "Basic OmJhcg=="


def test_authenticator_falls_back_to_keyring_url(
    config: Config,
    mock_remote: None,
    http: type[httpretty.httpretty],
    with_simple_keyring: None,
    dummy_keyring: DummyBackend,
):
    config.merge(
        {
            "repositories": {"foo": {"url": "https://foo.bar/simple/"}},
        }
    )

    dummy_keyring.set_password(
        "https://foo.bar/simple/", None, SimpleCredential(None, "bar")
    )

    authenticator = Authenticator(config, NullIO())
    authenticator.request("get", "https://foo.bar/files/foo-0.1.0.tar.gz")

    request = http.last_request()

    assert request.headers["Authorization"] == "Basic OmJhcg=="


def test_authenticator_falls_back_to_keyring_netloc(
    config: Config,
    mock_remote: None,
    http: type[httpretty.httpretty],
    with_simple_keyring: None,
    dummy_keyring: DummyBackend,
):
    config.merge(
        {
            "repositories": {"foo": {"url": "https://foo.bar/simple/"}},
        }
    )

    dummy_keyring.set_password("foo.bar", None, SimpleCredential(None, "bar"))

    authenticator = Authenticator(config, NullIO())
    authenticator.request("get", "https://foo.bar/files/foo-0.1.0.tar.gz")

    request = http.last_request()

    assert request.headers["Authorization"] == "Basic OmJhcg=="


@pytest.mark.filterwarnings("ignore::pytest.PytestUnhandledThreadExceptionWarning")
def test_authenticator_request_retries_on_exception(
    mocker: MockerFixture, config: Config, http: type[httpretty.httpretty]
):
    sleep = mocker.patch("time.sleep")
    sdist_uri = f"https://foo.bar/files/{uuid.uuid4()!s}/foo-0.1.0.tar.gz"
    content = str(uuid.uuid4())
    seen = []

    def callback(
        request: requests.Request, uri: str, response_headers: dict
    ) -> list[int | dict | str]:
        if seen.count(uri) < 2:
            seen.append(uri)
            raise requests.exceptions.ConnectionError("Disconnected")
        return [200, response_headers, content]

    httpretty.register_uri(httpretty.GET, sdist_uri, body=callback)

    authenticator = Authenticator(config, NullIO())
    response = authenticator.request("get", sdist_uri)
    assert response.text == content
    assert sleep.call_count == 2


@pytest.mark.filterwarnings("ignore::pytest.PytestUnhandledThreadExceptionWarning")
def test_authenticator_request_raises_exception_when_attempts_exhausted(
    mocker: MockerFixture, config: Config, http: type[httpretty.httpretty]
):
    sleep = mocker.patch("time.sleep")
    sdist_uri = f"https://foo.bar/files/{uuid.uuid4()!s}/foo-0.1.0.tar.gz"

    def callback(*_: Any, **___: Any) -> None:
        raise requests.exceptions.ConnectionError(str(uuid.uuid4()))

    httpretty.register_uri(httpretty.GET, sdist_uri, body=callback)
    authenticator = Authenticator(config, NullIO())

    with pytest.raises(requests.exceptions.ConnectionError):
        authenticator.request("get", sdist_uri)

    assert sleep.call_count == 5


@pytest.mark.parametrize(
    ["status", "attempts"],
    [
        (400, 0),
        (401, 0),
        (403, 0),
        (404, 0),
        (500, 0),
        (502, 5),
        (503, 5),
        (504, 5),
    ],
)
def test_authenticator_request_retries_on_status_code(
    mocker: MockerFixture,
    config: Config,
    http: type[httpretty.httpretty],
    status: int,
    attempts: int,
):
    sleep = mocker.patch("time.sleep")
    sdist_uri = f"https://foo.bar/files/{uuid.uuid4()!s}/foo-0.1.0.tar.gz"
    content = str(uuid.uuid4())

    def callback(
        request: requests.Request, uri: str, response_headers: dict
    ) -> list[int | dict | str]:
        return [status, response_headers, content]

    httpretty.register_uri(httpretty.GET, sdist_uri, body=callback)
    authenticator = Authenticator(config, NullIO())

    with pytest.raises(requests.exceptions.HTTPError) as excinfo:
        authenticator.request("get", sdist_uri)

    assert excinfo.value.response.status_code == status
    assert excinfo.value.response.text == content

    assert sleep.call_count == attempts


def test_authenticator_uses_env_provided_credentials(
    config: Config,
    environ: None,
    mock_remote: type[httpretty.httpretty],
    http: type[httpretty.httpretty],
    monkeypatch: MonkeyPatch,
):
    monkeypatch.setenv("POETRY_HTTP_BASIC_FOO_USERNAME", "bar")
    monkeypatch.setenv("POETRY_HTTP_BASIC_FOO_PASSWORD", "baz")

    config.merge({"repositories": {"foo": {"url": "https://foo.bar/simple/"}}})

    authenticator = Authenticator(config, NullIO())
    authenticator.request("get", "https://foo.bar/files/foo-0.1.0.tar.gz")

    request = http.last_request()

    assert request.headers["Authorization"] == "Basic YmFyOmJheg=="


@pytest.mark.parametrize(
    "cert,client_cert",
    [
        (None, None),
        (None, "path/to/provided/client-cert"),
        ("/path/to/provided/cert", None),
        ("/path/to/provided/cert", "path/to/provided/client-cert"),
    ],
)
def test_authenticator_uses_certs_from_config_if_not_provided(
    config: Config,
    mock_remote: type[httpretty.httpretty],
    http: type[httpretty.httpretty],
    mocker: MockerFixture,
    cert: str | None,
    client_cert: str | None,
):
    configured_cert = "/path/to/cert"
    configured_client_cert = "/path/to/client-cert"
    config.merge(
        {
            "repositories": {"foo": {"url": "https://foo.bar/simple/"}},
            "http-basic": {"foo": {"username": "bar", "password": "baz"}},
            "certificates": {
                "foo": {"cert": configured_cert, "client-cert": configured_client_cert}
            },
        }
    )

    authenticator = Authenticator(config, NullIO())
    url = "https://foo.bar/files/foo-0.1.0.tar.gz"
    session = authenticator.get_session(url)
    session_send = mocker.patch.object(session, "send")
    authenticator.request(
        "get",
        url,
        verify=cert,
        cert=client_cert,
    )
    kwargs = session_send.call_args[1]

    assert Path(kwargs["verify"]) == Path(cert or configured_cert)
    assert Path(kwargs["cert"]) == Path(client_cert or configured_client_cert)


def test_authenticator_uses_credentials_from_config_matched_by_url_path(
    config: Config, mock_remote: None, http: type[httpretty.httpretty]
):
    config.merge(
        {
            "repositories": {
                "foo-alpha": {"url": "https://foo.bar/alpha/files/simple/"},
                "foo-beta": {"url": "https://foo.bar/beta/files/simple/"},
            },
            "http-basic": {
                "foo-alpha": {"username": "bar", "password": "alpha"},
                "foo-beta": {"username": "baz", "password": "beta"},
            },
        }
    )

    authenticator = Authenticator(config, NullIO())
    authenticator.request("get", "https://foo.bar/alpha/files/simple/foo-0.1.0.tar.gz")

    request = http.last_request()

    basic_auth = base64.b64encode(b"bar:alpha").decode()
    assert request.headers["Authorization"] == f"Basic {basic_auth}"

    # Make request on second repository with the same netloc but different credentials
    authenticator.request("get", "https://foo.bar/beta/files/simple/foo-0.1.0.tar.gz")

    request = http.last_request()

    basic_auth = base64.b64encode(b"baz:beta").decode()
    assert request.headers["Authorization"] == f"Basic {basic_auth}"


def test_authenticator_uses_credentials_from_config_with_at_sign_in_path(
    config: Config, mock_remote: None, http: type[httpretty.httpretty]
):
    config.merge(
        {
            "repositories": {
                "foo": {"url": "https://foo.bar/beta/files/simple/"},
            },
            "http-basic": {
                "foo": {"username": "bar", "password": "baz"},
            },
        }
    )
    authenticator = Authenticator(config, NullIO())
    authenticator.request("get", "https://foo.bar/beta/files/simple/f@@-0.1.0.tar.gz")

    request = http.last_request()

    basic_auth = base64.b64encode(b"bar:baz").decode()
    assert request.headers["Authorization"] == f"Basic {basic_auth}"


def test_authenticator_falls_back_to_keyring_url_matched_by_path(
    config: Config,
    mock_remote: None,
    http: type[httpretty.httpretty],
    with_simple_keyring: None,
    dummy_keyring: DummyBackend,
):
    config.merge(
        {
            "repositories": {
                "foo-alpha": {"url": "https://foo.bar/alpha/files/simple/"},
                "foo-beta": {"url": "https://foo.bar/beta/files/simple/"},
            }
        }
    )

    dummy_keyring.set_password(
        "https://foo.bar/alpha/files/simple/", None, SimpleCredential(None, "bar")
    )
    dummy_keyring.set_password(
        "https://foo.bar/beta/files/simple/", None, SimpleCredential(None, "baz")
    )

    authenticator = Authenticator(config, NullIO())

    authenticator.request("get", "https://foo.bar/alpha/files/simple/foo-0.1.0.tar.gz")
    request = http.last_request()

    basic_auth = base64.b64encode(b":bar").decode()
    assert request.headers["Authorization"] == f"Basic {basic_auth}"

    authenticator.request("get", "https://foo.bar/beta/files/simple/foo-0.1.0.tar.gz")
    request = http.last_request()

    basic_auth = base64.b64encode(b":baz").decode()
    assert request.headers["Authorization"] == f"Basic {basic_auth}"


def test_authenticator_uses_env_provided_credentials_matched_by_url_path(
    config: Config,
    environ: None,
    mock_remote: type[httpretty.httpretty],
    http: type[httpretty.httpretty],
    monkeypatch: MonkeyPatch,
):
    monkeypatch.setenv("POETRY_HTTP_BASIC_FOO_ALPHA_USERNAME", "bar")
    monkeypatch.setenv("POETRY_HTTP_BASIC_FOO_ALPHA_PASSWORD", "alpha")
    monkeypatch.setenv("POETRY_HTTP_BASIC_FOO_BETA_USERNAME", "baz")
    monkeypatch.setenv("POETRY_HTTP_BASIC_FOO_BETA_PASSWORD", "beta")

    config.merge(
        {
            "repositories": {
                "foo-alpha": {"url": "https://foo.bar/alpha/files/simple/"},
                "foo-beta": {"url": "https://foo.bar/beta/files/simple/"},
            }
        }
    )

    authenticator = Authenticator(config, NullIO())

    authenticator.request("get", "https://foo.bar/alpha/files/simple/foo-0.1.0.tar.gz")
    request = http.last_request()

    basic_auth = base64.b64encode(b"bar:alpha").decode()
    assert request.headers["Authorization"] == f"Basic {basic_auth}"

    authenticator.request("get", "https://foo.bar/beta/files/simple/foo-0.1.0.tar.gz")
    request = http.last_request()

    basic_auth = base64.b64encode(b"baz:beta").decode()
    assert request.headers["Authorization"] == f"Basic {basic_auth}"


def test_authenticator_azure_feed_guid_credentials(
    config: Config,
    mock_remote: None,
    http: type[httpretty.httpretty],
    with_simple_keyring: None,
    dummy_keyring: DummyBackend,
):
    config.merge(
        {
            "repositories": {
                "alpha": {
                    "url": "https://foo.bar/org-alpha/_packaging/feed/pypi/simple/"
                },
                "beta": {
                    "url": "https://foo.bar/org-beta/_packaging/feed/pypi/simple/"
                },
            },
            "http-basic": {
                "alpha": {"username": "foo", "password": "bar"},
                "beta": {"username": "baz", "password": "qux"},
            },
        }
    )

    authenticator = Authenticator(config, NullIO())

    authenticator.request(
        "get",
        "https://foo.bar/org-alpha/_packaging/GUID/pypi/simple/a/1.0.0/a-1.0.0.whl",
    )
    request = http.last_request()

    basic_auth = base64.b64encode(b"foo:bar").decode()
    assert request.headers["Authorization"] == f"Basic {basic_auth}"

    authenticator.request(
        "get",
        "https://foo.bar/org-beta/_packaging/GUID/pypi/simple/b/1.0.0/a-1.0.0.whl",
    )
    request = http.last_request()

    basic_auth = base64.b64encode(b"baz:qux").decode()
    assert request.headers["Authorization"] == f"Basic {basic_auth}"


def test_authenticator_add_repository(
    config: Config,
    mock_remote: None,
    http: type[httpretty.httpretty],
    with_simple_keyring: None,
    dummy_keyring: DummyBackend,
):
    config.merge(
        {
            "http-basic": {
                "source": {"username": "foo", "password": "bar"},
            },
        }
    )

    authenticator = Authenticator(config, NullIO())

    authenticator.request(
        "get",
        "https://foo.bar/simple/a/1.0.0/a-1.0.0.whl",
    )
    request = http.last_request()
    assert "Authorization" not in request.headers

    authenticator.add_repository("source", "https://foo.bar/simple/")

    authenticator.request(
        "get",
        "https://foo.bar/simple/a/1.0.0/a-1.0.0.whl",
    )
    request = http.last_request()

    basic_auth = base64.b64encode(b"foo:bar").decode()
    assert request.headers["Authorization"] == f"Basic {basic_auth}"


def test_authenticator_git_repositories(
    config: Config,
    mock_remote: None,
    http: type[httpretty.httpretty],
    with_simple_keyring: None,
    dummy_keyring: DummyBackend,
):
    config.merge(
        {
            "repositories": {
                "one": {"url": "https://foo.bar/org/one.git"},
                "two": {"url": "https://foo.bar/org/two.git"},
            },
            "http-basic": {
                "one": {"username": "foo", "password": "bar"},
                "two": {"username": "baz", "password": "qux"},
            },
        }
    )

    authenticator = Authenticator(config, NullIO())

    one = authenticator.get_credentials_for_git_url("https://foo.bar/org/one.git")
    assert one.username == "foo"
    assert one.password == "bar"

    two = authenticator.get_credentials_for_git_url("https://foo.bar/org/two.git")
    assert two.username == "baz"
    assert two.password == "qux"

    two_ssh = authenticator.get_credentials_for_git_url("ssh://git@foo.bar/org/two.git")
    assert not two_ssh.username
    assert not two_ssh.password

    three = authenticator.get_credentials_for_git_url("https://foo.bar/org/three.git")
    assert not three.username
    assert not three.password
