mirror of
https://github.com/home-assistant/core.git
synced 2026-05-28 19:26:11 +01:00
Stefan Agner
481 lines
16 KiB
Python
481 lines
16 KiB
Python
"""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
|