199 lines
7.0 KiB
Python
199 lines
7.0 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
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="test-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="test-token"))
|
|
client._session.get = Mock(side_effect=requests.Timeout("timed out")) # 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()
|
|
|
|
|
|
def test_get_history_calls_home_assistant_history_api() -> None:
|
|
response = _response(payload=[[{"entity_id": "sensor.temperature", "state": "21.0"}]])
|
|
client = _client_with_response(response)
|
|
start = datetime(2026, 6, 1, tzinfo=timezone.utc)
|
|
end = datetime(2026, 6, 2, tzinfo=timezone.utc)
|
|
|
|
payload = client.get_history(["sensor.temperature"], start, end)
|
|
|
|
assert payload == [[{"entity_id": "sensor.temperature", "state": "21.0"}]]
|
|
client._session.get.assert_called_once() # type: ignore[attr-defined]
|
|
call = client._session.get.call_args # type: ignore[attr-defined]
|
|
assert "/api/history/period/2026-06-01T00:00:00+00:00" in call.args[0]
|
|
assert call.kwargs["params"]["filter_entity_id"] == "sensor.temperature"
|
|
assert call.kwargs["params"]["end_time"] == "2026-06-02T00:00:00+00:00"
|
|
|
|
|
|
def test_list_entity_metadata_calls_template_api() -> None:
|
|
response = _response()
|
|
response.text = (
|
|
'[{"entity_id":"sensor.temperature","area_name":"Kueche","device_name":"Thermometer"}]'
|
|
)
|
|
client = HaClient(HaClientSettings(url="http://ha.local", token="test-token"))
|
|
client._session.post = Mock(return_value=response) # type: ignore[method-assign]
|
|
|
|
metadata = client.list_entity_metadata(["sensor.temperature"])
|
|
|
|
assert metadata == {
|
|
"sensor.temperature": {
|
|
"area_id": None,
|
|
"area_name": "Kueche",
|
|
"device_id": None,
|
|
"device_name": "Thermometer",
|
|
}
|
|
}
|
|
|
|
|
|
def test_list_entity_metadata_batches_template_calls() -> None:
|
|
responses = []
|
|
for index in range(3):
|
|
response = _response()
|
|
response.text = (
|
|
f'[{{"entity_id":"sensor.test_{index}",'
|
|
f'"area_name":"Area {index}","device_name":"Device {index}"}}]'
|
|
)
|
|
responses.append(response)
|
|
client = HaClient(HaClientSettings(url="http://ha.local", token="test-token"))
|
|
client._session.post = Mock(side_effect=responses) # type: ignore[method-assign]
|
|
|
|
entity_ids = [f"sensor.test_{index}" for index in range(401)]
|
|
metadata = client.list_entity_metadata(entity_ids)
|
|
|
|
assert client._session.post.call_count == 3
|
|
assert metadata["sensor.test_0"]["area_name"] == "Area 0"
|
|
assert metadata["sensor.test_1"]["device_name"] == "Device 1"
|
|
assert metadata["sensor.test_2"]["device_name"] == "Device 2"
|
|
|
|
|
|
def test_get_logbook_filters_entity_and_period() -> None:
|
|
response = _response(payload=[{"entity_id": "light.office"}])
|
|
client = _client_with_response(response)
|
|
start = datetime(2026, 6, 1, tzinfo=timezone.utc)
|
|
end = datetime(2026, 6, 2, tzinfo=timezone.utc)
|
|
|
|
payload = client.get_logbook("light.office", start, end)
|
|
|
|
assert payload == [{"entity_id": "light.office"}]
|
|
call = client._session.get.call_args # type: ignore[attr-defined]
|
|
assert "/api/logbook/2026-06-01T00:00:00+00:00" in call.args[0]
|
|
assert call.kwargs["params"]["entity"] == "light.office"
|
|
|
|
|
|
def test_call_service_posts_to_home_assistant() -> None:
|
|
response = _response(payload=[])
|
|
client = HaClient(HaClientSettings(url="http://ha.local", token="test-token"))
|
|
client._session.post = Mock(return_value=response) # type: ignore[method-assign]
|
|
|
|
result = client.call_service("light", "turn_on", {"entity_id": "light.office"})
|
|
|
|
assert result == []
|
|
client._session.post.assert_called_once_with(
|
|
"http://ha.local/api/services/light/turn_on",
|
|
json={"entity_id": "light.office"},
|
|
timeout=10,
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("entity_ids", "start", "end"),
|
|
[
|
|
(
|
|
[],
|
|
datetime(2026, 6, 1, tzinfo=timezone.utc),
|
|
datetime(2026, 6, 2, tzinfo=timezone.utc),
|
|
),
|
|
(
|
|
["sensor.temperature"],
|
|
datetime(2026, 6, 1),
|
|
datetime(2026, 6, 2, tzinfo=timezone.utc),
|
|
),
|
|
(
|
|
["sensor.temperature"],
|
|
datetime(2026, 6, 2, tzinfo=timezone.utc),
|
|
datetime(2026, 6, 1, tzinfo=timezone.utc),
|
|
),
|
|
(
|
|
["invalid entity"],
|
|
datetime(2026, 6, 1, tzinfo=timezone.utc),
|
|
datetime(2026, 6, 2, tzinfo=timezone.utc),
|
|
),
|
|
(
|
|
["sensor.temperature"],
|
|
datetime(2026, 5, 1, tzinfo=timezone.utc),
|
|
datetime(2026, 6, 2, tzinfo=timezone.utc),
|
|
),
|
|
],
|
|
)
|
|
def test_get_history_validates_request(
|
|
entity_ids: list[str],
|
|
start: datetime,
|
|
end: datetime,
|
|
) -> None:
|
|
client = HaClient(HaClientSettings(url="http://ha.local", token="test-token"))
|
|
with pytest.raises(ValueError):
|
|
client.get_history(entity_ids, start, end)
|