"""The tests for the hassio component."""

from http import HTTPStatus

from aiohttp import StreamReader
from aiohttp.test_utils import TestClient
import pytest

from tests.common import MockUser
from tests.test_util.aiohttp import AiohttpClientMocker


@pytest.fixture
def hassio_user_client(
    hassio_client: TestClient, hass_admin_user: MockUser
) -> TestClient:
    """Return a Hass.io HTTP client tied to a non-admin user."""
    hass_admin_user.groups = []
    return hassio_client


@pytest.mark.parametrize(
    "path",
    [
        "addons/bl_b392/logo",
        "addons/bl_b392/icon",
    ],
)
async def test_forward_request_onboarded_user_get(
    hassio_user_client: TestClient, aioclient_mock: AiohttpClientMocker, path: str
) -> None:
    """Test fetching normal path."""
    aioclient_mock.get(f"http://127.0.0.1/{path}", text="response")

    resp = await hassio_user_client.get(f"/api/hassio/{path}")

    # Check we got right response
    assert resp.status == HTTPStatus.OK
    body = await resp.text()
    assert body == "response"

    # Check we forwarded command
    assert len(aioclient_mock.mock_calls) == 1
    # We only expect a single header.
    assert aioclient_mock.mock_calls[0][3] == {"X-Hass-Source": "core.http"}


@pytest.mark.parametrize("method", ["POST", "PUT", "DELETE"])
async def test_forward_request_onboarded_user_unallowed_methods(
    hassio_user_client: TestClient, aioclient_mock: AiohttpClientMocker, method: str
) -> None:
    """Test fetching normal path."""
    resp = await hassio_user_client.request(method, "/api/hassio/addons/bl_b392/icon")

    # Check we got right response
    assert resp.status == HTTPStatus.METHOD_NOT_ALLOWED

    # Check we did not forward command
    assert len(aioclient_mock.mock_calls) == 0


@pytest.mark.parametrize(
    ("bad_path", "expected_status"),
    [
        # Caught by bullshit filter
        ("addons/bl_b392/%252E./icon", HTTPStatus.BAD_REQUEST),
        # Unauthenticated path
        ("supervisor/info", HTTPStatus.UNAUTHORIZED),
        ("supervisor/logs", HTTPStatus.UNAUTHORIZED),
        ("supervisor/logs/follow", HTTPStatus.UNAUTHORIZED),
        ("addons/bl_b392/logs", HTTPStatus.UNAUTHORIZED),
        ("addons/bl_b392/logs/follow", HTTPStatus.UNAUTHORIZED),
    ],
)
async def test_forward_request_onboarded_user_unallowed_paths(
    hassio_user_client: TestClient,
    aioclient_mock: AiohttpClientMocker,
    bad_path: str,
    expected_status: int,
) -> None:
    """Test fetching normal path."""
    resp = await hassio_user_client.get(f"/api/hassio/{bad_path}")

    # Check we got right response
    assert resp.status == expected_status
    # Check we didn't forward command
    assert len(aioclient_mock.mock_calls) == 0


@pytest.mark.parametrize(
    "path",
    [
        "addons/bl_b392/logo",
        "addons/bl_b392/icon",
    ],
)
async def test_forward_request_onboarded_noauth_get(
    hassio_noauth_client: TestClient, aioclient_mock: AiohttpClientMocker, path: str
) -> None:
    """Test fetching normal path."""
    aioclient_mock.get(f"http://127.0.0.1/{path}", text="response")

    resp = await hassio_noauth_client.get(f"/api/hassio/{path}")

    # Check we got right response
    assert resp.status == HTTPStatus.OK
    body = await resp.text()
    assert body == "response"

    # Check we forwarded command
    assert len(aioclient_mock.mock_calls) == 1
    # We only expect a single header.
    assert aioclient_mock.mock_calls[0][3] == {"X-Hass-Source": "core.http"}


@pytest.mark.parametrize("method", ["POST", "PUT", "DELETE"])
async def test_forward_request_onboarded_noauth_unallowed_methods(
    hassio_noauth_client: TestClient, aioclient_mock: AiohttpClientMocker, method: str
) -> None:
    """Test fetching normal path."""
    resp = await hassio_noauth_client.request(method, "/api/hassio/addons/bl_b392/icon")

    # Check we got right response
    assert resp.status == HTTPStatus.METHOD_NOT_ALLOWED

    # Check we did not forward command
    assert len(aioclient_mock.mock_calls) == 0


@pytest.mark.parametrize(
    ("bad_path", "expected_status"),
    [
        # Caught by bullshit filter
        ("addons/bl_b392/%252E./icon", HTTPStatus.BAD_REQUEST),
        # Unauthenticated path
        ("supervisor/info", HTTPStatus.UNAUTHORIZED),
        ("supervisor/logs", HTTPStatus.UNAUTHORIZED),
        ("supervisor/logs/follow", HTTPStatus.UNAUTHORIZED),
        ("addons/bl_b392/logs", HTTPStatus.UNAUTHORIZED),
        ("addons/bl_b392/logs/follow", HTTPStatus.UNAUTHORIZED),
    ],
)
async def test_forward_request_onboarded_noauth_unallowed_paths(
    hassio_noauth_client: TestClient,
    aioclient_mock: AiohttpClientMocker,
    bad_path: str,
    expected_status: int,
) -> None:
    """Test fetching normal path."""
    resp = await hassio_noauth_client.get(f"/api/hassio/{bad_path}")

    # Check we got right response
    assert resp.status == expected_status
    # Check we didn't forward command
    assert len(aioclient_mock.mock_calls) == 0


BACKUP_NON_ADMIN_REJECTED_PARAMS = [
    ("GET", "backups/1234abcd/info"),
    ("GET", "backups/1234abcd/download"),
    ("POST", "backups/1234abcd/restore/full"),
    ("POST", "backups/1234abcd/restore/partial"),
    ("POST", "backups/new/upload"),
]


@pytest.mark.parametrize(("method", "path"), BACKUP_NON_ADMIN_REJECTED_PARAMS)
async def test_forward_request_backup_unauthenticated_rejected(
    hassio_noauth_client: TestClient,
    aioclient_mock: AiohttpClientMocker,
    method: str,
    path: str,
) -> None:
    """Test backup endpoints reject unauthenticated requests."""
    resp = await hassio_noauth_client.request(method, f"/api/hassio/{path}")

    assert resp.status == HTTPStatus.UNAUTHORIZED
    assert len(aioclient_mock.mock_calls) == 0


@pytest.mark.parametrize(("method", "path"), BACKUP_NON_ADMIN_REJECTED_PARAMS)
async def test_forward_request_backup_non_admin_rejected(
    hassio_user_client: TestClient,
    aioclient_mock: AiohttpClientMocker,
    method: str,
    path: str,
) -> None:
    """Test backup endpoints reject authenticated non-admin requests."""
    resp = await hassio_user_client.request(method, f"/api/hassio/{path}")

    assert resp.status == HTTPStatus.UNAUTHORIZED
    assert len(aioclient_mock.mock_calls) == 0


@pytest.mark.parametrize(
    ("path", "authenticated"),
    [
        ("addons/bl_b392/logo", False),
        ("addons/bl_b392/icon", False),
        ("backups/1234abcd/info", True),
        ("supervisor/logs", True),
        ("supervisor/logs/follow", True),
        ("addons/bl_b392/logs", True),
        ("addons/bl_b392/logs/follow", True),
        ("addons/bl_b392/changelog", True),
        ("addons/bl_b392/documentation", True),
    ],
)
async def test_forward_request_admin_get(
    hassio_client: TestClient,
    aioclient_mock: AiohttpClientMocker,
    path: str,
    authenticated: bool,
) -> None:
    """Test fetching normal path."""
    aioclient_mock.get(f"http://127.0.0.1/{path}", text="response")

    resp = await hassio_client.get(f"/api/hassio/{path}")

    # Check we got right response
    assert resp.status == HTTPStatus.OK
    body = await resp.text()
    assert body == "response"

    # Check we forwarded command
    assert len(aioclient_mock.mock_calls) == 1
    expected_headers = {
        "X-Hass-Source": "core.http",
    }
    if authenticated:
        expected_headers["Authorization"] = "Bearer 123456"

    assert aioclient_mock.mock_calls[0][3] == expected_headers


@pytest.mark.parametrize(
    "path",
    [
        "backups/new/upload",
        "backups/1234abcd/restore/full",
        "backups/1234abcd/restore/partial",
    ],
)
async def test_forward_request_admin_post(
    hassio_client: TestClient,
    aioclient_mock: AiohttpClientMocker,
    path: str,
) -> None:
    """Test fetching normal path."""
    aioclient_mock.get(f"http://127.0.0.1/{path}", text="response")

    resp = await hassio_client.get(f"/api/hassio/{path}")

    # Check we got right response
    assert resp.status == HTTPStatus.OK
    body = await resp.text()
    assert body == "response"

    # Check we forwarded command
    assert len(aioclient_mock.mock_calls) == 1
    # We only expect a single header.
    assert aioclient_mock.mock_calls[0][3] == {
        "X-Hass-Source": "core.http",
        "Authorization": "Bearer 123456",
    }


@pytest.mark.parametrize("method", ["POST", "PUT", "DELETE"])
async def test_forward_request_admin_unallowed_methods(
    hassio_client: TestClient, aioclient_mock: AiohttpClientMocker, method: str
) -> None:
    """Test fetching normal path."""
    resp = await hassio_client.request(method, "/api/hassio/addons/bl_b392/icon")

    # Check we got right response
    assert resp.status == HTTPStatus.METHOD_NOT_ALLOWED

    # Check we did not forward command
    assert len(aioclient_mock.mock_calls) == 0


@pytest.mark.parametrize(
    ("bad_path", "expected_status"),
    [
        # Caught by bullshit filter
        ("addons/bl_b392/%252E./icon", HTTPStatus.BAD_REQUEST),
        # Unauthenticated path
        ("supervisor/info", HTTPStatus.UNAUTHORIZED),
    ],
)
async def test_forward_request_admin_unallowed_paths(
    hassio_client: TestClient,
    aioclient_mock: AiohttpClientMocker,
    bad_path: str,
    expected_status: int,
) -> None:
    """Test fetching normal path."""
    resp = await hassio_client.get(f"/api/hassio/{bad_path}")

    # Check we got right response
    assert resp.status == expected_status
    # Check we didn't forward command
    assert len(aioclient_mock.mock_calls) == 0


async def test_bad_gateway_when_cannot_find_supervisor(
    hassio_client: TestClient, aioclient_mock: AiohttpClientMocker
) -> None:
    """Test we get a bad gateway error if we can't find supervisor."""
    aioclient_mock.get("http://127.0.0.1/addons/bl_b392/icon", exc=TimeoutError)

    resp = await hassio_client.get("/api/hassio/addons/bl_b392/icon")
    assert resp.status == HTTPStatus.BAD_GATEWAY


async def test_backup_upload_headers(
    hassio_client: TestClient,
    aioclient_mock: AiohttpClientMocker,
) -> None:
    """Test that we forward the full header for backup upload."""
    content_type = "multipart/form-data; boundary='--webkit'"
    aioclient_mock.post("http://127.0.0.1/backups/new/upload")

    resp = await hassio_client.post(
        "/api/hassio/backups/new/upload", headers={"Content-Type": content_type}
    )

    # Check we got right response
    assert resp.status == HTTPStatus.OK

    assert len(aioclient_mock.mock_calls) == 1

    req_headers = aioclient_mock.mock_calls[0][-1]
    assert req_headers["Content-Type"] == content_type


async def test_backup_download_headers(
    hassio_client: TestClient, aioclient_mock: AiohttpClientMocker
) -> None:
    """Test that we forward the full header for backup download."""
    content_disposition = "attachment; filename=test.tar"
    aioclient_mock.get(
        "http://127.0.0.1/backups/1234abcd/download",
        headers={
            "Content-Length": "50000000",
            "Content-Disposition": content_disposition,
        },
    )

    resp = await hassio_client.get("/api/hassio/backups/1234abcd/download")

    # Check we got right response
    assert resp.status == HTTPStatus.OK

    assert len(aioclient_mock.mock_calls) == 1

    assert resp.headers["Content-Disposition"] == content_disposition


async def test_stream(
    hassio_client: TestClient, aioclient_mock: AiohttpClientMocker
) -> None:
    """Verify that the request is a stream."""
    content_type = "multipart/form-data; boundary='--webkit'"
    aioclient_mock.post("http://127.0.0.1/backups/new/upload")
    resp = await hassio_client.post(
        "/api/hassio/backups/new/upload", headers={"Content-Type": content_type}
    )
    # Check we got right response
    assert resp.status == HTTPStatus.OK
    assert isinstance(aioclient_mock.mock_calls[-1][2], StreamReader)


async def test_simple_get_no_stream(
    hassio_client: TestClient, aioclient_mock: AiohttpClientMocker
) -> None:
    """Verify that a simple GET request is not a stream."""
    aioclient_mock.get("http://127.0.0.1/addons/bl_b392/icon")
    resp = await hassio_client.get("/api/hassio/addons/bl_b392/icon")
    assert resp.status == HTTPStatus.OK
    assert aioclient_mock.mock_calls[-1][2] is None


async def test_no_follow_logs_compress(
    hassio_client: TestClient, aioclient_mock: AiohttpClientMocker
) -> None:
    """Test that we do not compress follow logs."""
    aioclient_mock.get(
        "http://127.0.0.1/supervisor/logs/follow",
        headers={"Content-Type": "text/plain"},
    )
    aioclient_mock.get(
        "http://127.0.0.1/supervisor/logs",
        headers={"Content-Type": "text/plain"},
    )

    resp1 = await hassio_client.get("/api/hassio/supervisor/logs/follow")
    resp2 = await hassio_client.get("/api/hassio/supervisor/logs")

    # Check we got right response
    assert resp1.status == HTTPStatus.OK
    assert resp1.headers.get("Content-Encoding") is None

    assert resp2.status == HTTPStatus.OK
    assert resp2.headers.get("Content-Encoding") == "deflate"


async def test_no_event_stream_compress(
    hassio_client: TestClient, aioclient_mock: AiohttpClientMocker
) -> None:
    """Test that we do not compress SSE (Server-Sent Events) streams."""
    aioclient_mock.get(
        "http://127.0.0.1/backups/1234abcd/info",
        headers={"Content-Type": "text/event-stream"},
    )
    aioclient_mock.get(
        "http://127.0.0.1/addons/bl_b392/changelog",
        headers={"Content-Type": "application/json"},
    )

    resp1 = await hassio_client.get("/api/hassio/backups/1234abcd/info")
    resp2 = await hassio_client.get("/api/hassio/addons/bl_b392/changelog")

    # Check we got right response
    assert resp1.status == HTTPStatus.OK
    # SSE (text/event-stream) should not be compressed to allow streaming
    assert resp1.headers.get("Content-Encoding") is None

    assert resp2.status == HTTPStatus.OK
    # Regular JSON should be compressed
    assert resp2.headers.get("Content-Encoding") == "deflate"


async def test_forward_range_header_for_logs(
    hassio_client: TestClient, aioclient_mock: AiohttpClientMocker
) -> None:
    """Test that we forward the Range header for logs."""
    aioclient_mock.get("http://127.0.0.1/host/logs")
    aioclient_mock.get("http://127.0.0.1/host/logs/boots/-1")
    aioclient_mock.get("http://127.0.0.1/host/logs/boots/-2/follow?lines=100")
    aioclient_mock.get("http://127.0.0.1/addons/123abc_esphome/logs")
    aioclient_mock.get("http://127.0.0.1/addons/123abc_esphome/logs/follow")
    aioclient_mock.get("http://127.0.0.1/backups/1234abcd/download")

    test_range = ":-100:50"

    host_resp = await hassio_client.get(
        "/api/hassio/host/logs", headers={"Range": test_range}
    )
    host_resp2 = await hassio_client.get(
        "/api/hassio/host/logs/boots/-1", headers={"Range": test_range}
    )
    host_resp3 = await hassio_client.get(
        "/api/hassio/host/logs/boots/-2/follow?lines=100", headers={"Range": test_range}
    )
    addon_resp = await hassio_client.get(
        "/api/hassio/addons/123abc_esphome/logs", headers={"Range": test_range}
    )
    addon_resp2 = await hassio_client.get(
        "/api/hassio/addons/123abc_esphome/logs/follow", headers={"Range": test_range}
    )
    backup_resp = await hassio_client.get(
        "/api/hassio/backups/1234abcd/download", headers={"Range": test_range}
    )

    assert host_resp.status == HTTPStatus.OK
    assert host_resp2.status == HTTPStatus.OK
    assert host_resp3.status == HTTPStatus.OK
    assert addon_resp.status == HTTPStatus.OK
    assert addon_resp2.status == HTTPStatus.OK
    assert backup_resp.status == HTTPStatus.OK

    assert len(aioclient_mock.mock_calls) == 6

    assert aioclient_mock.mock_calls[0][-1].get("Range") == test_range
    assert aioclient_mock.mock_calls[1][-1].get("Range") == test_range
    assert aioclient_mock.mock_calls[2][-1].get("Range") == test_range
    assert aioclient_mock.mock_calls[3][-1].get("Range") == test_range
    assert aioclient_mock.mock_calls[4][-1].get("Range") == test_range
    assert aioclient_mock.mock_calls[5][-1].get("Range") is None