"""The tests for the rest command platform."""

import base64
from http import HTTPStatus
from unittest.mock import patch

import aiohttp
from multidict import CIMultiDict
import pytest
from yarl import URL

from homeassistant.components.rest_command import DOMAIN
from homeassistant.const import (
    CONTENT_TYPE_JSON,
    CONTENT_TYPE_TEXT_PLAIN,
    HTTP_DIGEST_AUTHENTICATION,
    SERVICE_RELOAD,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError

from .conftest import TEST_URL, ComponentSetup

from tests.test_util.aiohttp import AiohttpClientMocker


async def test_reload(hass: HomeAssistant, setup_component: ComponentSetup) -> None:
    """Verify we can reload rest_command integration."""
    await setup_component()

    assert hass.services.has_service(DOMAIN, "get_test")
    assert not hass.services.has_service(DOMAIN, "new_test")

    new_config = {
        DOMAIN: {
            "new_test": {"url": "https://example.org", "method": "get"},
        }
    }
    with patch(
        "homeassistant.config.load_yaml_config_file",
        autospec=True,
        return_value=new_config,
    ):
        await hass.services.async_call(DOMAIN, SERVICE_RELOAD, blocking=True)

    assert hass.services.has_service(DOMAIN, "new_test")
    assert not hass.services.has_service(DOMAIN, "get_test")


async def test_setup_tests(
    hass: HomeAssistant, setup_component: ComponentSetup
) -> None:
    """Set up test config and test it."""
    await setup_component()

    assert hass.services.has_service(DOMAIN, "get_test")
    assert hass.services.has_service(DOMAIN, "post_test")
    assert hass.services.has_service(DOMAIN, "put_test")
    assert hass.services.has_service(DOMAIN, "delete_test")


async def test_rest_command_timeout(
    hass: HomeAssistant,
    setup_component: ComponentSetup,
    aioclient_mock: AiohttpClientMocker,
) -> None:
    """Call a rest command with timeout."""
    await setup_component()

    aioclient_mock.get(TEST_URL, exc=TimeoutError())

    with pytest.raises(HomeAssistantError) as exc:
        await hass.services.async_call(DOMAIN, "get_test", {}, blocking=True)
    assert str(exc.value) == 'Timeout when calling resource "https://example.com/"'

    assert len(aioclient_mock.mock_calls) == 1


async def test_rest_command_aiohttp_error(
    hass: HomeAssistant,
    setup_component: ComponentSetup,
    aioclient_mock: AiohttpClientMocker,
) -> None:
    """Call a rest command with aiohttp exception."""
    await setup_component()

    aioclient_mock.get(TEST_URL, exc=aiohttp.ClientError())

    with pytest.raises(HomeAssistantError) as exc:
        await hass.services.async_call(DOMAIN, "get_test", {}, blocking=True)

    assert (
        str(exc.value)
        == 'Client error occurred when calling resource "https://example.com/"'
    )
    assert len(aioclient_mock.mock_calls) == 1


async def test_rest_command_http_error(
    hass: HomeAssistant,
    setup_component: ComponentSetup,
    aioclient_mock: AiohttpClientMocker,
) -> None:
    """Call a rest command with status code 400."""
    await setup_component()

    aioclient_mock.get(TEST_URL, status=HTTPStatus.BAD_REQUEST)

    await hass.services.async_call(DOMAIN, "get_test", {}, blocking=True)

    assert len(aioclient_mock.mock_calls) == 1


async def test_rest_command_auth(
    hass: HomeAssistant,
    setup_component: ComponentSetup,
    aioclient_mock: AiohttpClientMocker,
) -> None:
    """Call a rest command with auth credential."""
    await setup_component()

    aioclient_mock.get(TEST_URL, content=b"success")

    await hass.services.async_call(DOMAIN, "auth_test", {}, blocking=True)

    assert len(aioclient_mock.mock_calls) == 1


async def test_rest_command_digest_auth(
    hass: HomeAssistant,
    setup_component: ComponentSetup,
    aioclient_mock: AiohttpClientMocker,
) -> None:
    """Call a rest command with HTTP digest authentication."""
    config = {
        "digest_auth_test": {
            "url": TEST_URL,
            "method": "get",
            "username": "test_user",
            "password": "test_pass",
            "authentication": HTTP_DIGEST_AUTHENTICATION,
        }
    }

    await setup_component(config)

    # Mock the digest auth behavior - the request will be called
    # with DigestAuthMiddleware
    with patch("aiohttp.ClientSession.get") as mock_get:

        async def async_iter_chunks(self, chunk_size):
            yield b"success"

        mock_response = type(
            "MockResponse",
            (),
            {
                "status": 200,
                "content_type": "text/plain",
                "headers": {},
                "url": TEST_URL,
                "content": type(
                    "MockContent", (), {"iter_chunked": async_iter_chunks}
                )(),
            },
        )()
        mock_get.return_value.__aenter__.return_value = mock_response

        await hass.services.async_call(DOMAIN, "digest_auth_test", {}, blocking=True)

        # Verify that the request was made with DigestAuthMiddleware
        assert mock_get.called
        call_kwargs = mock_get.call_args[1]
        assert "middlewares" in call_kwargs
        assert len(call_kwargs["middlewares"]) == 1
        assert isinstance(call_kwargs["middlewares"][0], aiohttp.DigestAuthMiddleware)


async def test_rest_command_form_data(
    hass: HomeAssistant,
    setup_component: ComponentSetup,
    aioclient_mock: AiohttpClientMocker,
) -> None:
    """Call a rest command with post form data."""
    await setup_component()

    aioclient_mock.post(TEST_URL, content=b"success")

    await hass.services.async_call(DOMAIN, "post_test", {}, blocking=True)

    assert len(aioclient_mock.mock_calls) == 1
    assert aioclient_mock.mock_calls[0][2] == b"test"


@pytest.mark.parametrize(
    "method",
    [
        "get",
        "patch",
        "post",
        "put",
        "delete",
    ],
)
async def test_rest_command_methods(
    hass: HomeAssistant,
    setup_component: ComponentSetup,
    aioclient_mock: AiohttpClientMocker,
    method: str,
) -> None:
    """Test various http methods."""
    await setup_component()

    aioclient_mock.request(method=method, url=TEST_URL, content=b"success")

    await hass.services.async_call(DOMAIN, f"{method}_test", {}, blocking=True)

    assert len(aioclient_mock.mock_calls) == 1


async def test_rest_command_headers(
    hass: HomeAssistant,
    setup_component: ComponentSetup,
    aioclient_mock: AiohttpClientMocker,
) -> None:
    """Call a rest command with custom headers and content types."""
    header_config_variations = {
        "no_headers_test": {},
        "content_type_test": {"content_type": CONTENT_TYPE_TEXT_PLAIN},
        "headers_test": {
            "headers": {
                "Accept": CONTENT_TYPE_JSON,
                "User-Agent": "Mozilla/5.0",
            }
        },
        "headers_and_content_type_test": {
            "headers": {"Accept": CONTENT_TYPE_JSON},
            "content_type": CONTENT_TYPE_TEXT_PLAIN,
        },
        "headers_and_content_type_override_test": {
            "headers": {
                "Accept": CONTENT_TYPE_JSON,
                aiohttp.hdrs.CONTENT_TYPE: "application/pdf",
            },
            "content_type": CONTENT_TYPE_TEXT_PLAIN,
        },
        "headers_template_test": {
            "headers": {
                "Accept": CONTENT_TYPE_JSON,
                "User-Agent": "Mozilla/{{ 3 + 2 }}.0",
            }
        },
        "headers_and_content_type_override_template_test": {
            "headers": {
                "Accept": "application/{{ 1 + 1 }}json",
                aiohttp.hdrs.CONTENT_TYPE: "application/pdf",
            },
            "content_type": "text/json",
        },
    }

    # add common parameters
    for variation in header_config_variations.values():
        variation.update({"url": TEST_URL, "method": "post", "payload": "test data"})

    await setup_component(header_config_variations)

    # provide post request data
    aioclient_mock.post(TEST_URL, content=b"success")

    for test_service in (
        "no_headers_test",
        "content_type_test",
        "headers_test",
        "headers_and_content_type_test",
        "headers_and_content_type_override_test",
        "headers_template_test",
        "headers_and_content_type_override_template_test",
    ):
        await hass.services.async_call(DOMAIN, test_service, {}, blocking=True)

    await hass.async_block_till_done()
    assert len(aioclient_mock.mock_calls) == 7

    # no_headers_test
    assert aioclient_mock.mock_calls[0][3] is None

    # content_type_test
    assert len(aioclient_mock.mock_calls[1][3]) == 1
    assert (
        aioclient_mock.mock_calls[1][3].get(aiohttp.hdrs.CONTENT_TYPE)
        == CONTENT_TYPE_TEXT_PLAIN
    )

    # headers_test
    assert len(aioclient_mock.mock_calls[2][3]) == 2
    assert aioclient_mock.mock_calls[2][3].get("Accept") == CONTENT_TYPE_JSON
    assert aioclient_mock.mock_calls[2][3].get("User-Agent") == "Mozilla/5.0"

    # headers_and_content_type_test
    assert len(aioclient_mock.mock_calls[3][3]) == 2
    assert (
        aioclient_mock.mock_calls[3][3].get(aiohttp.hdrs.CONTENT_TYPE)
        == CONTENT_TYPE_TEXT_PLAIN
    )
    assert aioclient_mock.mock_calls[3][3].get("Accept") == CONTENT_TYPE_JSON

    # headers_and_content_type_override_test
    assert len(aioclient_mock.mock_calls[4][3]) == 2
    assert (
        aioclient_mock.mock_calls[4][3].get(aiohttp.hdrs.CONTENT_TYPE)
        == CONTENT_TYPE_TEXT_PLAIN
    )
    assert aioclient_mock.mock_calls[4][3].get("Accept") == CONTENT_TYPE_JSON

    # headers_template_test
    assert len(aioclient_mock.mock_calls[5][3]) == 2
    assert aioclient_mock.mock_calls[5][3].get("Accept") == CONTENT_TYPE_JSON
    assert aioclient_mock.mock_calls[5][3].get("User-Agent") == "Mozilla/5.0"

    # headers_and_content_type_override_template_test
    assert len(aioclient_mock.mock_calls[6][3]) == 2
    assert aioclient_mock.mock_calls[6][3].get(aiohttp.hdrs.CONTENT_TYPE) == "text/json"
    assert aioclient_mock.mock_calls[6][3].get("Accept") == "application/2json"


async def test_rest_command_get_response_plaintext(
    hass: HomeAssistant,
    setup_component: ComponentSetup,
    aioclient_mock: AiohttpClientMocker,
) -> None:
    """Get rest_command response, text."""
    await setup_component()

    aioclient_mock.get(
        TEST_URL, content=b"success", headers={"content-type": "text/plain"}
    )

    response = await hass.services.async_call(
        DOMAIN, "get_test", {}, blocking=True, return_response=True
    )

    assert len(aioclient_mock.mock_calls) == 1
    assert response["content"] == "success"
    assert response["status"] == 200
    assert response["headers"] == {"content-type": "text/plain"}


async def test_rest_command_get_response_json(
    hass: HomeAssistant,
    setup_component: ComponentSetup,
    aioclient_mock: AiohttpClientMocker,
) -> None:
    """Get rest_command response, json."""
    await setup_component()

    aioclient_mock.get(
        TEST_URL,
        json={"status": "success", "number": 42},
        headers={"content-type": "application/json"},
    )

    response = await hass.services.async_call(
        DOMAIN, "get_test", {}, blocking=True, return_response=True
    )

    assert len(aioclient_mock.mock_calls) == 1
    assert response["content"]["status"] == "success"
    assert response["content"]["number"] == 42
    assert response["status"] == 200
    assert response["headers"] == {"content-type": "application/json"}


async def test_rest_command_get_response_multiple_headers(
    hass: HomeAssistant,
    setup_component: ComponentSetup,
    aioclient_mock: AiohttpClientMocker,
) -> None:
    """Get rest_command response with multiple headers of the same name."""
    await setup_component()

    aioclient_mock.get(
        TEST_URL,
        content=b"success",
        headers=CIMultiDict(
            [
                ("content-type", "text/plain"),
                ("set-cookie", "foo=bar; Path=/"),
                ("set-cookie", "baz=qux; Path=/"),
            ]
        ),
    )

    response = await hass.services.async_call(
        DOMAIN, "get_test", {}, blocking=True, return_response=True
    )

    assert len(aioclient_mock.mock_calls) == 1
    assert response["content"] == "success"
    assert response["status"] == 200
    assert response["headers"] == {
        "content-type": "text/plain",
        "set-cookie": ["foo=bar; Path=/", "baz=qux; Path=/"],
    }


async def test_rest_command_get_response_malformed_json(
    hass: HomeAssistant,
    setup_component: ComponentSetup,
    aioclient_mock: AiohttpClientMocker,
) -> None:
    """Get rest_command response, malformed json."""
    await setup_component()

    aioclient_mock.get(
        TEST_URL,
        content=b'{"status": "failure", 42',
        headers={"content-type": "application/json"},
    )

    # No problem without 'return_response'
    response = await hass.services.async_call(DOMAIN, "get_test", {}, blocking=True)
    assert not response

    # Throws error when requesting response
    with pytest.raises(HomeAssistantError) as exc:
        await hass.services.async_call(
            DOMAIN, "get_test", {}, blocking=True, return_response=True
        )
    assert (
        str(exc.value)
        == 'The response of "https://example.com/" could not be decoded as JSON'
    )


async def test_rest_command_get_response_none(
    hass: HomeAssistant,
    setup_component: ComponentSetup,
    aioclient_mock: AiohttpClientMocker,
) -> None:
    """Get rest_command response, other."""
    await setup_component()

    png = base64.decodebytes(
        b"iVBORw0KGgoAAAANSUhEUgAAAAIAAAABCAIAAAB7QOjdAAAAAXNSR0IArs4c6QAAAARnQU1BAACxjwv8YQ"
        b"UAAAAJcEhZcwAAFiUAABYlAUlSJPAAAAAPSURBVBhXY/h/ku////8AECAE1JZPvDAAAAAASUVORK5CYII="
    )

    aioclient_mock.get(
        TEST_URL,
        content=png,
        headers={"content-type": "text/plain"},
    )

    # No problem without 'return_response'
    response = await hass.services.async_call(DOMAIN, "get_test", {}, blocking=True)
    assert not response

    # Throws Decode error when requesting response
    with pytest.raises(HomeAssistantError) as exc:
        response = await hass.services.async_call(
            DOMAIN, "get_test", {}, blocking=True, return_response=True
        )
    assert (
        str(exc.value)
        == 'The response of "https://example.com/" could not be decoded as text'
    )

    assert not response


async def test_rest_command_response_iter_chunked(
    hass: HomeAssistant,
    setup_component: ComponentSetup,
    aioclient_mock: AiohttpClientMocker,
) -> None:
    """Ensure response is consumed when return_response is False."""
    await setup_component()

    png = base64.decodebytes(
        b"iVBORw0KGgoAAAANSUhEUgAAAAIAAAABCAIAAAB7QOjdAAAAAXNSR0IArs4c6QAAAARnQU1BAACxjwv8YQ"
        b"UAAAAJcEhZcwAAFiUAABYlAUlSJPAAAAAPSURBVBhXY/h/ku////8AECAE1JZPvDAAAAAASUVORK5CYII="
    )
    aioclient_mock.get(TEST_URL, content=png)

    with patch("aiohttp.StreamReader.iter_chunked", autospec=True) as mock_iter_chunked:
        response = await hass.services.async_call(DOMAIN, "get_test", {}, blocking=True)

        # Ensure the response is not returned
        assert response is None

        # Verify iter_chunked was called with a chunk size
        assert mock_iter_chunked.called


async def test_rest_command_skip_url_encoding(
    hass: HomeAssistant,
    setup_component: ComponentSetup,
    aioclient_mock: AiohttpClientMocker,
) -> None:
    """Check URL encoding."""
    config = {
        "skip_url_encoding_test": {
            "url": "0%2C",
            "method": "get",
            "skip_url_encoding": True,
        },
        "with_url_encoding_test": {
            "url": "1,",
            "method": "get",
        },
    }

    await setup_component(config)

    aioclient_mock.get(URL("0%2C", encoded=True), content=b"success")
    aioclient_mock.get(URL("1,"), content=b"success")

    await hass.services.async_call(DOMAIN, "skip_url_encoding_test", {}, blocking=True)
    await hass.services.async_call(DOMAIN, "with_url_encoding_test", {}, blocking=True)

    assert len(aioclient_mock.mock_calls) == 2
    assert str(aioclient_mock.mock_calls[0][1]) == "0%2C"
    assert str(aioclient_mock.mock_calls[1][1]) == "1,"