147 lines
4.5 KiB
Python
147 lines
4.5 KiB
Python
from collections.abc import Sequence
|
|
from datetime import datetime
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.ha.exceptions import HaTimeoutError
|
|
from app.ha.discovery import DiscoveredEntity, EntityRole
|
|
from app.ha.history import EntityHistorySeries, NumericHistoryPoint
|
|
from app.ha.models import HaEntitySummary
|
|
from app.ha.reader import HaReader
|
|
from app.main import app
|
|
|
|
|
|
class FakeHaReader(HaReader):
|
|
def __init__(self) -> None:
|
|
pass
|
|
|
|
def read_entities(self) -> Sequence[HaEntitySummary]:
|
|
return [HaEntitySummary(entity_id="sensor.temperature", domain="sensor")]
|
|
|
|
def discover(
|
|
self,
|
|
domains: set[str] | None = None,
|
|
learnable: bool | None = None,
|
|
) -> Sequence[DiscoveredEntity]:
|
|
result = DiscoveredEntity(
|
|
entity_id="sensor.temperature",
|
|
domain="sensor",
|
|
device_class="temperature",
|
|
category="temperature",
|
|
role=EntityRole.MEASUREMENT,
|
|
learnable=True,
|
|
reason="Numerischer Messsensor für Zeitreihen und Training.",
|
|
)
|
|
if domains and result.domain not in domains:
|
|
return []
|
|
if learnable is not None and result.learnable is not learnable:
|
|
return []
|
|
return [result]
|
|
|
|
def read_history(
|
|
self,
|
|
entity_ids: list[str],
|
|
start_time: datetime,
|
|
end_time: datetime,
|
|
) -> Sequence[EntityHistorySeries]:
|
|
return [
|
|
EntityHistorySeries(
|
|
entity_id=entity_ids[0],
|
|
points=[NumericHistoryPoint(timestamp=start_time, value=21.5)],
|
|
)
|
|
]
|
|
|
|
|
|
class TimeoutHaReader(HaReader):
|
|
def __init__(self) -> None:
|
|
pass
|
|
|
|
def read_entities(self) -> Sequence[HaEntitySummary]:
|
|
raise HaTimeoutError("contains internal details that must not leak")
|
|
|
|
|
|
def test_openapi_docs_are_available() -> None:
|
|
with TestClient(app) as client:
|
|
response = client.get("/docs")
|
|
assert response.status_code == 200
|
|
assert "SillyHome Next API" in response.text
|
|
|
|
|
|
def test_entities_returns_reader_data() -> None:
|
|
with TestClient(app) as client:
|
|
app.state.ha_reader = FakeHaReader()
|
|
response = client.get("/v1/entities")
|
|
assert response.status_code == 200
|
|
assert response.json() == [
|
|
{
|
|
"entity_id": "sensor.temperature",
|
|
"domain": "sensor",
|
|
"state": None,
|
|
"last_changed": None,
|
|
"state_class": None,
|
|
"device_class": None,
|
|
"unit_of_measurement": None,
|
|
"friendly_name": None,
|
|
"area_id": None,
|
|
"area_name": None,
|
|
"device_id": None,
|
|
"device_name": None,
|
|
}
|
|
]
|
|
|
|
|
|
def test_entities_returns_503_without_home_assistant_config() -> None:
|
|
with TestClient(app) as client:
|
|
response = client.get("/v1/entities")
|
|
assert response.status_code == 503
|
|
|
|
|
|
def test_entities_maps_ha_errors_without_leaking_details() -> None:
|
|
with TestClient(app) as client:
|
|
app.state.ha_reader = TimeoutHaReader()
|
|
response = client.get("/v1/entities")
|
|
assert response.status_code == 504
|
|
assert response.json() == {"detail": "Home Assistant request timed out."}
|
|
|
|
|
|
def test_discovery_filters_entities() -> None:
|
|
with TestClient(app) as client:
|
|
app.state.ha_reader = FakeHaReader()
|
|
response = client.get("/v1/discovery?domain=sensor&learnable=true")
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == [
|
|
{
|
|
"entity_id": "sensor.temperature",
|
|
"domain": "sensor",
|
|
"device_class": "temperature",
|
|
"state_class": None,
|
|
"unit_of_measurement": None,
|
|
"category": "temperature",
|
|
"role": "measurement",
|
|
"learnable": True,
|
|
"reason": "Numerischer Messsensor für Zeitreihen und Training.",
|
|
}
|
|
]
|
|
|
|
|
|
def test_history_returns_normalized_series() -> None:
|
|
with TestClient(app) as client:
|
|
app.state.ha_reader = FakeHaReader()
|
|
response = client.get(
|
|
"/v1/history",
|
|
params=[
|
|
("entity_id", "sensor.temperature"),
|
|
("start_time", "2026-06-01T00:00:00Z"),
|
|
("end_time", "2026-06-02T00:00:00Z"),
|
|
],
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == [
|
|
{
|
|
"entity_id": "sensor.temperature",
|
|
"points": [{"timestamp": "2026-06-01T00:00:00Z", "value": 21.5}],
|
|
}
|
|
]
|