from __future__ import annotations

import getpass
import logging
from unittest import mock

import pytest
import requests.sessions

import jira.client
from jira.exceptions import JIRAError, NotJIRAInstanceError
from tests.conftest import JiraTestManager, get_unique_project_name


@pytest.fixture()
def prep():
    pass


@pytest.fixture(scope="module")
def test_manager() -> JiraTestManager:
    return JiraTestManager()


@pytest.fixture()
def cl_admin(test_manager: JiraTestManager) -> jira.client.JIRA:
    return test_manager.jira_admin


@pytest.fixture()
def cl_normal(test_manager: JiraTestManager) -> jira.client.JIRA:
    return test_manager.jira_normal


@pytest.fixture(scope="function")
def slug(request: pytest.FixtureRequest, cl_admin: jira.client.JIRA):
    """Project slug."""
    slug = get_unique_project_name()
    project_name = f"Test user={getpass.getuser()} key={slug} A"

    try:
        proj = cl_admin.project(slug)
    except JIRAError:
        proj = cl_admin.create_project(slug, project_name)
    assert proj

    yield slug
    try:
        cl_admin.delete_project(slug, enable_undo=False)
    except (ValueError, JIRAError):
        # Some tests have project already removed, so we stay silent
        pass


@pytest.fixture(scope="session")
def stream_logger():
    logger = logging.getLogger("test_logger")
    logger.addHandler(logging.StreamHandler())
    return logger


@pytest.fixture(scope="session")
def mock_not_jira_client(stream_logger):
    class MockClient:
        def __init__(self, is_cloud=False):
            self.is_cloud = is_cloud
            self.log = stream_logger

        @property
        def _is_cloud(self):
            return self.is_cloud

        @jira.client.cloud_api
        def mock_cloud_only_method(self, *args, **kwargs):
            return args, kwargs

        @jira.client.experimental_atlassian_api
        def mock_experimental_method(self, *args, **kwargs):
            return args, kwargs

        @jira.client.experimental_atlassian_api
        def mock_method_raises_jira_error(self, *args, **kwargs):
            raise JIRAError(**kwargs)

    return MockClient


@pytest.fixture(scope="session")
def mock_jira_client(stream_logger):
    class MockClient(jira.client.JIRA):
        def __init__(self, is_cloud=False):
            self.is_cloud = is_cloud
            self.log = stream_logger

        @property
        def _is_cloud(self):
            return self.is_cloud

        @jira.client.cloud_api
        def mock_cloud_only_method(self, *args, **kwargs):
            return args, kwargs

        @jira.client.experimental_atlassian_api
        def mock_experimental_method(self, *args, **kwargs):
            return args, kwargs

        @jira.client.experimental_atlassian_api
        def mock_method_raises_jira_error(self, *args, **kwargs):
            raise JIRAError(**kwargs)

    return MockClient


@pytest.fixture(scope="session")
def mock_response():
    class MockResponse:
        def __init__(self, status_code=404):
            self.status_code = status_code
            self.url = "some/url/that/does/not/exist"

    return MockResponse


def test_delete_project(cl_admin, cl_normal, slug):
    assert cl_admin.delete_project(slug)


def test_delete_inexistent_project(cl_admin):
    slug = "abogus123"
    with pytest.raises(JIRAError) as ex:
        assert cl_admin.delete_project(slug)

    assert "No project could be found with key" in str(
        ex.value
    ) or f'Parameter pid="{slug}" is not a Project, projectID or slug' in str(ex.value)


def test_templates(cl_admin):
    templates = set(cl_admin.templates())
    expected_templates = set(
        filter(
            None,
            """
Basic software development
Kanban software development
Process management
Project management
Scrum software development
Task management
""".split("\n"),
        )
    )

    assert templates == expected_templates


def test_result_list():
    iterable = [2, 3]
    startAt = 0
    maxResults = 50
    total = 2

    results = jira.client.ResultList(iterable, startAt, maxResults, total)

    for idx, result in enumerate(results):
        assert results[idx] == iterable[idx]

    assert next(results) == iterable[0]
    assert next(results) == iterable[1]

    with pytest.raises(StopIteration):
        next(results)


def test_result_list_if_empty():
    results = jira.client.ResultList()

    for r in results:
        raise AssertionError("`results` should be empty")

    with pytest.raises(StopIteration):
        next(results)


@pytest.mark.parametrize(
    "options_arg",
    [
        {"headers": {"Content-Type": "application/json;charset=UTF-8"}},
        {"headers": {"random-header": "nice random"}},
    ],
    ids=["overwrite", "new"],
)
def test_headers_unclobbered_update(options_arg, no_fields):
    assert "headers" in options_arg, "test case options must contain headers"

    # GIVEN: the headers and the expected value
    header_to_check: str = list(options_arg["headers"].keys())[0]
    expected_header_value: str = options_arg["headers"][header_to_check]

    invariant_header_name: str = "X-Atlassian-Token"
    invariant_header_value: str = jira.client.JIRA.DEFAULT_OPTIONS["headers"][
        invariant_header_name
    ]

    # We arbitrarily chose a header to check it remains unchanged/unclobbered
    # so should not be overwritten by a test case
    assert (
        invariant_header_name not in options_arg["headers"]
    ), f"{invariant_header_name} is checked as not being overwritten in this test"

    # WHEN: we initialise the JIRA class and get the headers
    jira_client = jira.client.JIRA(
        server="https://jira.atlasian.com",
        get_server_info=False,
        validate=False,
        options=options_arg,
    )

    session_headers = jira_client._session.headers

    # THEN: we have set the right headers and not affect the other headers' defaults
    assert session_headers[header_to_check] == expected_header_value
    assert session_headers[invariant_header_name] == invariant_header_value


def test_headers_unclobbered_update_with_no_provided_headers(no_fields):
    options_arg = {}  # a dict with "headers" not set

    # GIVEN:the headers and the expected value
    invariant_header_name: str = "X-Atlassian-Token"
    invariant_header_value: str = jira.client.JIRA.DEFAULT_OPTIONS["headers"][
        invariant_header_name
    ]

    # WHEN: we initialise the JIRA class with no provided headers and get the headers
    jira_client = jira.client.JIRA(
        server="https://jira.atlasian.com",
        get_server_info=False,
        validate=False,
        options=options_arg,
    )

    session_headers = jira_client._session.headers

    # THEN: we have not affected the other headers' defaults
    assert session_headers[invariant_header_name] == invariant_header_value


def test_token_auth(cl_admin: jira.client.JIRA):
    """Tests the Personal Access Token authentication works."""
    # GIVEN: We have a PAT token created by a user.
    pat_token_request = {
        "name": "my_new_token",
        "expirationDuration": 1,
    }
    base_url = cl_admin.server_url
    pat_token_response = cl_admin._session.post(
        f"{base_url}/rest/pat/latest/tokens", json=pat_token_request
    ).json()
    new_token = pat_token_response["rawToken"]

    # WHEN: A new client is authenticated with this token
    new_jira_client = jira.client.JIRA(token_auth=new_token)

    # THEN: The reported authenticated user of the token
    # matches the original token creator user.
    assert cl_admin.myself() == new_jira_client.myself()


def test_bearer_token_auth():
    my_token = "cool-token"
    token_auth_jira = jira.client.JIRA(
        server="https://what.ever",
        token_auth=my_token,
        get_server_info=False,
        validate=False,
    )
    method_send = token_auth_jira._session.send
    with mock.patch.object(token_auth_jira._session, method_send.__name__) as mock_send:
        token_auth_jira._session.get(token_auth_jira.server_url)
        prepared_req: requests.sessions.PreparedRequest = mock_send.call_args[0][0]
        assert prepared_req.headers["Authorization"] == f"Bearer {my_token}"


def test_cookie_auth(test_manager: JiraTestManager):
    """Test Cookie based authentication works.

    NOTE: this is deprecated in Cloud and is not recommended in Server.
    https://developer.atlassian.com/cloud/jira/platform/jira-rest-api-cookie-based-authentication/
    https://developer.atlassian.com/server/jira/platform/cookie-based-authentication/
    """
    # GIVEN: the username and password
    # WHEN: We create a session with cookie auth for the same server
    cookie_auth_jira = jira.client.JIRA(
        server=test_manager.CI_JIRA_URL,
        auth=(test_manager.CI_JIRA_ADMIN, test_manager.CI_JIRA_ADMIN_PASSWORD),
    )
    # THEN: We get the same result from the API
    assert test_manager.jira_admin.myself() == cookie_auth_jira.myself()


def test_cookie_auth_retry():
    """Test Cookie based authentication retry logic works."""
    # GIVEN: arguments that will cause a 401 error
    auth_class = jira.client.JiraCookieAuth
    reset_func = jira.client.JiraCookieAuth._reset_401_retry_counter
    new_options = jira.client.JIRA.DEFAULT_OPTIONS.copy()
    new_options["auth_url"] = "/401"
    with pytest.raises(JIRAError):
        with mock.patch.object(auth_class, reset_func.__name__) as mock_reset_func:
            # WHEN: We create a session with cookie auth
            jira.client.JIRA(
                server="https://httpstat.us",
                options=new_options,
                auth=("user", "pass"),
            )
    # THEN: We don't get a RecursionError and only call the reset_function once
    mock_reset_func.assert_called_once()


def test_createmeta_issuetypes_pagination(cl_normal, slug):
    """Test createmeta_issuetypes pagination kwargs"""
    issue_types_resp = cl_normal.createmeta_issuetypes(slug, startAt=50, maxResults=100)
    assert issue_types_resp["startAt"] == 50
    assert issue_types_resp["maxResults"] == 100


def test_createmeta_fieldtypes_pagination(cl_normal, slug):
    """Test createmeta_fieldtypes pagination kwargs"""
    issue_types = cl_normal.createmeta_issuetypes(slug)
    assert issue_types["total"]
    issue_type_id = issue_types["values"][-1]["id"]
    field_types_resp = cl_normal.createmeta_fieldtypes(
        projectIdOrKey=slug, issueTypeId=issue_type_id, startAt=50, maxResults=100
    )
    assert field_types_resp["startAt"] == 50
    assert field_types_resp["maxResults"] == 100


@pytest.mark.parametrize(
    "mock_client_method", ["mock_cloud_only_method", "mock_experimental_method"]
)
def test_not_cloud_instance(mock_not_jira_client, mock_client_method):
    client = mock_not_jira_client()
    with pytest.raises(NotJIRAInstanceError) as exc:
        getattr(client, mock_client_method)()

    assert str(exc.value) == (
        "The first argument of this function must be an instance of type "
        f"JIRA. Instance Type: {mock_not_jira_client().__class__.__name__}"
    )


@mock.patch("requests.Session.request")
def test_cloud_api(mock_request, mock_jira_client):
    mock_client = mock_jira_client(is_cloud=True)
    out = mock_client.mock_cloud_only_method("one", two="three")
    assert out is not None


@mock.patch("requests.Session.request")
def test_cloud_api_not_cloud_server(mock_request, mock_jira_client, caplog):
    mock_client = mock_jira_client()
    mock_client.mock_cloud_only_method()
    assert caplog.messages[0] == (
        "This functionality is not available on Jira Data Center (Server) version."
    )


@mock.patch("requests.Session.request")
def test_experimental(mock_request, mock_jira_client):
    out = mock_jira_client().mock_experimental_method("one", two="three")
    assert out is not None


@pytest.mark.parametrize("http_status_code", [404, 405])
@mock.patch("requests.Session.request")
def test_experimental_missing_or_not_allowed(
    mock_request, mock_jira_client, mock_response, http_status_code, caplog
):
    mock_response = mock_response(status_code=http_status_code)
    response = mock_jira_client().mock_method_raises_jira_error(
        response=mock_response,
        request=mock_response,
        status_code=mock_response.status_code,
    )
    assert response is None
    assert caplog.messages[0] == (
        f"Functionality at path {mock_response.url} is/was experimental. Status Code: "
        f"{mock_response.status_code}"
    )


@mock.patch("requests.Session.request")
def test_experimental_non_200_not_404_405(
    mock_request, mock_jira_client, mock_response
):
    status_code = 400
    mock_response = mock_response(status_code=status_code)

    with pytest.raises(JIRAError) as ex:
        mock_jira_client().mock_method_raises_jira_error(
            response=mock_response,
            request=mock_response,
            status_code=mock_response.status_code,
        )

    assert ex.value.status_code == status_code
    assert isinstance(ex.value, JIRAError)