80 lines
2.4 KiB
Python
80 lines
2.4 KiB
Python
from __future__ import annotations
|
|
|
|
from unittest.mock import Mock
|
|
|
|
import pytest
|
|
import requests
|
|
|
|
from app.ha.client import HaClient, HaClientSettings
|
|
from app.ha.exceptions import (
|
|
HaAuthError,
|
|
HaHttpError,
|
|
HaTimeoutError,
|
|
HaUnexpectedPayloadError,
|
|
)
|
|
|
|
|
|
def _client_with_response(response: Mock) -> HaClient:
|
|
client = HaClient(HaClientSettings(url="http://ha.local", token="secret-token"))
|
|
client._session.get = Mock(return_value=response) # type: ignore[method-assign]
|
|
return client
|
|
|
|
|
|
def _response(status_code: int = 200, payload: object | None = None) -> Mock:
|
|
response = Mock()
|
|
response.status_code = status_code
|
|
response.json.return_value = [] if payload is None else payload
|
|
if status_code >= 400:
|
|
response.raise_for_status.side_effect = requests.HTTPError("upstream failed")
|
|
return response
|
|
|
|
|
|
def test_list_entities_returns_home_assistant_payload() -> None:
|
|
payload = [{"entity_id": "sensor.temperature", "state": "21"}]
|
|
client = _client_with_response(_response(payload=payload))
|
|
|
|
assert client.list_entities() == payload
|
|
|
|
|
|
def test_list_entities_maps_timeout() -> None:
|
|
client = HaClient(HaClientSettings(url="http://ha.local", token="secret-token"))
|
|
client._session.get = Mock(side_effect=requests.Timeout("secret-token")) # type: ignore[method-assign]
|
|
|
|
with pytest.raises(HaTimeoutError):
|
|
client.list_entities()
|
|
|
|
|
|
@pytest.mark.parametrize("status_code", [401, 403])
|
|
def test_list_entities_maps_auth_errors(status_code: int) -> None:
|
|
client = _client_with_response(_response(status_code=status_code))
|
|
|
|
with pytest.raises(HaAuthError) as exc_info:
|
|
client.list_entities()
|
|
|
|
assert exc_info.value.status_code == status_code
|
|
|
|
|
|
def test_list_entities_maps_http_errors() -> None:
|
|
client = _client_with_response(_response(status_code=500))
|
|
|
|
with pytest.raises(HaHttpError) as exc_info:
|
|
client.list_entities()
|
|
|
|
assert exc_info.value.status_code == 500
|
|
|
|
|
|
def test_list_entities_rejects_invalid_json() -> None:
|
|
response = _response()
|
|
response.json.side_effect = ValueError("not json")
|
|
client = _client_with_response(response)
|
|
|
|
with pytest.raises(HaUnexpectedPayloadError):
|
|
client.list_entities()
|
|
|
|
|
|
def test_list_entities_rejects_non_list_payload() -> None:
|
|
client = _client_with_response(_response(payload={"entity_id": "sensor.temperature"}))
|
|
|
|
with pytest.raises(HaUnexpectedPayloadError):
|
|
client.list_entities()
|