Compare commits
1 Commits
otto/main-
...
feature/do
| Author | SHA1 | Date | |
|---|---|---|---|
| f4f23796d5 |
@@ -1,3 +1,3 @@
|
||||
# Copy to .env for local development. Do not commit real tokens.
|
||||
SILLYHOME_HA_URL=http://homeassistant.local:8123
|
||||
SILLYHOME_HA_TOKEN=replace-with-a-long-lived-access-token
|
||||
# Home Assistant Zugriff
|
||||
SILLYHOME_HA_URL=http://localhost:8123
|
||||
SILLYHOME_HA_TOKEN=dein_long_lived_access_token
|
||||
@@ -1,31 +0,0 @@
|
||||
name: Quality
|
||||
|
||||
on:
|
||||
push:
|
||||
branches:
|
||||
- "**"
|
||||
pull_request:
|
||||
|
||||
jobs:
|
||||
test:
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- name: Checkout
|
||||
uses: actions/checkout@v4
|
||||
|
||||
- name: Set up Python
|
||||
uses: actions/setup-python@v5
|
||||
with:
|
||||
python-version: "3.11"
|
||||
|
||||
- name: Install project
|
||||
run: python -m pip install --upgrade pip && python -m pip install -e ".[dev]"
|
||||
|
||||
- name: Run tests
|
||||
run: pytest -q
|
||||
|
||||
- name: Run Ruff
|
||||
run: ruff check .
|
||||
|
||||
- name: Run Mypy
|
||||
run: mypy app tests
|
||||
1
.gitignore
vendored
1
.gitignore
vendored
@@ -10,4 +10,3 @@ __pycache__/
|
||||
.env
|
||||
.env.local
|
||||
.env.*
|
||||
!.env.example
|
||||
|
||||
69
README.md
69
README.md
@@ -13,52 +13,33 @@ TheSillyHome zeigte die Idee: statt statischer Regeln das Zuhause aus Verhaltens
|
||||
- Lokal-first ohne Cloudpflicht
|
||||
- Erweiterbar, testbar, dokumentiert
|
||||
|
||||
## Lokaler Quickstart
|
||||
## Quickstart (lokaler Betrieb)
|
||||
|
||||
Voraussetzung ist Python 3.11 oder neuer.
|
||||
1. **Voraussetzungen**
|
||||
- Python 3.11+
|
||||
- Home Assistant mit REST-API erreichbar
|
||||
- `pip install -e .[dev]`
|
||||
|
||||
```bash
|
||||
python -m venv .venv
|
||||
. .venv/bin/activate
|
||||
python -m pip install --upgrade pip
|
||||
python -m pip install -e ".[dev]"
|
||||
cp .env.example .env
|
||||
```
|
||||
2. **Umgebungsvariablen** (`.env` im Projektroot)
|
||||
```
|
||||
SILLYHOME_HA_URL=http://localhost:8123
|
||||
SILLYHOME_HA_TOKEN=dein_long_lived_access_token
|
||||
```
|
||||
Tipp: `.env.example` kopieren und anpassen. Tokens niemals committen!
|
||||
|
||||
In `.env` müssen für echte Home-Assistant-Daten diese Werte gesetzt werden:
|
||||
3. **Server starten**
|
||||
```
|
||||
uvicorn app.main:app --reload
|
||||
```
|
||||
|
||||
```bash
|
||||
SILLYHOME_HA_URL=http://homeassistant.local:8123
|
||||
SILLYHOME_HA_TOKEN=<long-lived-access-token>
|
||||
```
|
||||
4. **Prüfen**
|
||||
- OpenAPI-Docs: http://localhost:8000/docs
|
||||
- Health: http://localhost:8000/health
|
||||
- Entities: http://localhost:8000/v1/entities (benötigt gültige HA-Konfiguration)
|
||||
|
||||
Alternativ werden aus Kompatibilitätsgründen auch `HA_URL` und `HA_TOKEN` gelesen.
|
||||
Tokens bleiben lokal und dürfen nicht committed, geloggt oder in Issues kopiert werden.
|
||||
|
||||
API starten:
|
||||
|
||||
```bash
|
||||
uvicorn app.main:app --reload
|
||||
```
|
||||
|
||||
Nützliche Checks:
|
||||
|
||||
```bash
|
||||
curl http://127.0.0.1:8000/health
|
||||
curl http://127.0.0.1:8000/v1/entities
|
||||
```
|
||||
|
||||
Die interaktive API-Dokumentation liegt unter `http://127.0.0.1:8000/docs`.
|
||||
|
||||
## Qualität
|
||||
|
||||
Vor jedem Pull Request lokal laufen lassen:
|
||||
|
||||
```bash
|
||||
pytest -q
|
||||
ruff check .
|
||||
mypy app tests
|
||||
```
|
||||
|
||||
Der Gitea-Actions-Workflow in `.gitea/workflows/quality.yml` führt dieselben Checks für
|
||||
Pushes und Pull Requests aus.
|
||||
5. **Tests**
|
||||
```
|
||||
pytest -q
|
||||
ruff check .
|
||||
mypy app tests
|
||||
```
|
||||
|
||||
@@ -1 +0,0 @@
|
||||
"""SillyHome Next application package."""
|
||||
@@ -1 +0,0 @@
|
||||
"""API package."""
|
||||
@@ -1 +0,0 @@
|
||||
"""Version 1 API package."""
|
||||
@@ -1,12 +1,10 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Sequence
|
||||
from typing import List, Sequence
|
||||
|
||||
from fastapi import APIRouter, Depends
|
||||
from fastapi import APIRouter
|
||||
|
||||
from app.dependencies import get_ha_reader
|
||||
from app.ha.models import HaEntitySummary
|
||||
from app.ha.reader import HaReader
|
||||
|
||||
router = APIRouter(prefix="/v1", tags=["entities"])
|
||||
|
||||
@@ -15,7 +13,7 @@ router = APIRouter(prefix="/v1", tags=["entities"])
|
||||
"/entities",
|
||||
summary="Home-Assistant-Entities auflisten",
|
||||
description="Gibt eine kompakte Zusammenfassung aller erreichbaren HA-Entitäten zurück.",
|
||||
response_model=list[HaEntitySummary],
|
||||
response_model=List[HaEntitySummary],
|
||||
)
|
||||
def list_entities(reader: HaReader = Depends(get_ha_reader)) -> Sequence[HaEntitySummary]:
|
||||
return reader.read_entities()
|
||||
def list_entities() -> Sequence[HaEntitySummary]:
|
||||
raise NotImplementedError("Integration mit dem HA-Client folgt in separatem Issue.")
|
||||
|
||||
@@ -1,21 +0,0 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from dataclasses import dataclass
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Settings:
|
||||
ha_url: str | None = None
|
||||
ha_token: str | None = None
|
||||
|
||||
@property
|
||||
def ha_configured(self) -> bool:
|
||||
return bool(self.ha_url and self.ha_token)
|
||||
|
||||
|
||||
def load_settings() -> Settings:
|
||||
return Settings(
|
||||
ha_url=os.getenv("SILLYHOME_HA_URL") or os.getenv("HA_URL"),
|
||||
ha_token=os.getenv("SILLYHOME_HA_TOKEN") or os.getenv("HA_TOKEN"),
|
||||
)
|
||||
@@ -1 +0,0 @@
|
||||
"""Core application helpers."""
|
||||
@@ -1,23 +0,0 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from fastapi import FastAPI, Request, status
|
||||
from fastapi.responses import JSONResponse
|
||||
|
||||
from app.ha.exceptions import HaAuthError, HaClientError, HaHttpError, HaTimeoutError
|
||||
|
||||
|
||||
def register_exception_handlers(app: FastAPI) -> None:
|
||||
@app.exception_handler(HaClientError)
|
||||
async def handle_ha_client_error(_: Request, exc: HaClientError) -> JSONResponse:
|
||||
return JSONResponse(
|
||||
status_code=_status_code_for_ha_error(exc),
|
||||
content={"detail": exc.public_detail},
|
||||
)
|
||||
|
||||
|
||||
def _status_code_for_ha_error(exc: HaClientError) -> int:
|
||||
if isinstance(exc, HaTimeoutError):
|
||||
return status.HTTP_504_GATEWAY_TIMEOUT
|
||||
if isinstance(exc, (HaAuthError, HaHttpError)):
|
||||
return status.HTTP_502_BAD_GATEWAY
|
||||
return status.HTTP_502_BAD_GATEWAY
|
||||
@@ -1,15 +0,0 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from fastapi import HTTPException, Request, status
|
||||
|
||||
from app.ha.reader import HaReader
|
||||
|
||||
|
||||
def get_ha_reader(request: Request) -> HaReader:
|
||||
reader = getattr(request.app.state, "ha_reader", None)
|
||||
if not isinstance(reader, HaReader):
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
||||
detail="Home Assistant is not configured.",
|
||||
)
|
||||
return reader
|
||||
@@ -2,17 +2,9 @@ from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from dataclasses import dataclass
|
||||
from typing import Any
|
||||
|
||||
import requests
|
||||
|
||||
from app.ha.exceptions import (
|
||||
HaAuthError,
|
||||
HaHttpError,
|
||||
HaTimeoutError,
|
||||
HaUnexpectedPayloadError,
|
||||
)
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@@ -32,36 +24,10 @@ class HaClient:
|
||||
"Content-Type": "application/json",
|
||||
})
|
||||
|
||||
def list_entities(self) -> list[dict[str, Any]]:
|
||||
try:
|
||||
response = self._session.get(
|
||||
f"{self._settings.url}/api/states",
|
||||
timeout=self._settings.timeout_seconds,
|
||||
)
|
||||
except requests.Timeout as exc:
|
||||
raise HaTimeoutError("Home Assistant request timed out.") from exc
|
||||
except requests.RequestException as exc:
|
||||
raise HaHttpError(status_code=502, message="Home Assistant request failed.") from exc
|
||||
|
||||
if response.status_code in {401, 403}:
|
||||
raise HaAuthError(
|
||||
status_code=response.status_code,
|
||||
message="Home Assistant authentication failed.",
|
||||
)
|
||||
|
||||
try:
|
||||
response.raise_for_status()
|
||||
except requests.HTTPError as exc:
|
||||
raise HaHttpError(
|
||||
status_code=response.status_code,
|
||||
message="Home Assistant returned an HTTP error.",
|
||||
) from exc
|
||||
|
||||
try:
|
||||
payload = response.json()
|
||||
except ValueError as exc:
|
||||
raise HaUnexpectedPayloadError("Home Assistant returned invalid JSON.") from exc
|
||||
|
||||
if not isinstance(payload, list):
|
||||
raise HaUnexpectedPayloadError("Home Assistant states response must be a list.")
|
||||
return payload
|
||||
def list_entities(self) -> list[dict[str, object]]:
|
||||
response = self._session.get(
|
||||
f"{self._settings.url}/api/states",
|
||||
timeout=self._settings.timeout_seconds,
|
||||
)
|
||||
response.raise_for_status()
|
||||
return response.json()
|
||||
|
||||
@@ -1,27 +0,0 @@
|
||||
from __future__ import annotations
|
||||
|
||||
|
||||
class HaClientError(Exception):
|
||||
"""Base class for Home Assistant integration failures."""
|
||||
|
||||
public_detail = "Home Assistant is currently unavailable."
|
||||
|
||||
|
||||
class HaTimeoutError(HaClientError):
|
||||
public_detail = "Home Assistant request timed out."
|
||||
|
||||
|
||||
class HaHttpError(HaClientError):
|
||||
public_detail = "Home Assistant returned an error."
|
||||
|
||||
def __init__(self, status_code: int, message: str | None = None) -> None:
|
||||
super().__init__(message or self.public_detail)
|
||||
self.status_code = status_code
|
||||
|
||||
|
||||
class HaAuthError(HaHttpError):
|
||||
public_detail = "Home Assistant authentication failed."
|
||||
|
||||
|
||||
class HaUnexpectedPayloadError(HaClientError):
|
||||
public_detail = "Home Assistant returned an unexpected response."
|
||||
@@ -1,10 +1,9 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Sequence
|
||||
from typing import Any
|
||||
|
||||
from app.ha.client import HaClient
|
||||
from app.ha.models import HaEntitySummary
|
||||
from app.ha.models import HaEntitySummary, HaState
|
||||
|
||||
|
||||
class HaReader:
|
||||
@@ -19,21 +18,14 @@ class HaReader:
|
||||
if "." not in entity_id:
|
||||
continue
|
||||
domain = entity_id.split(".", 1)[0]
|
||||
raw_attributes = item.get("attributes") or {}
|
||||
attributes: dict[str, Any] = raw_attributes if isinstance(raw_attributes, dict) else {}
|
||||
attributes = item.get("attributes") or {}
|
||||
summaries.append(
|
||||
HaEntitySummary(
|
||||
entity_id=entity_id,
|
||||
domain=domain,
|
||||
state_class=_optional_str(attributes.get("state_class")),
|
||||
device_class=_optional_str(attributes.get("device_class")),
|
||||
unit_of_measurement=_optional_str(attributes.get("unit_of_measurement")),
|
||||
state_class=str(attributes.get("state_class") or ""),
|
||||
device_class=str(attributes.get("device_class") or ""),
|
||||
unit_of_measurement=str(attributes.get("unit_of_measurement") or ""),
|
||||
)
|
||||
)
|
||||
return summaries
|
||||
|
||||
|
||||
def _optional_str(value: object) -> str | None:
|
||||
if value is None or value == "":
|
||||
return None
|
||||
return str(value)
|
||||
|
||||
40
app/main.py
40
app/main.py
@@ -1,27 +1,20 @@
|
||||
from collections.abc import AsyncIterator
|
||||
from contextlib import asynccontextmanager
|
||||
|
||||
from fastapi import FastAPI
|
||||
from fastapi import FastAPI, Depends
|
||||
|
||||
from app.api.v1.entities import router as entities_router
|
||||
from app.config import load_settings
|
||||
from app.core.exception_handlers import register_exception_handlers
|
||||
from app.ha.client import HaClient, HaClientSettings
|
||||
from app.ha.reader import HaReader
|
||||
|
||||
|
||||
@asynccontextmanager
|
||||
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
|
||||
settings = load_settings()
|
||||
app.state.settings = settings
|
||||
if settings.ha_configured:
|
||||
client = HaClient(
|
||||
settings=HaClientSettings(
|
||||
url=settings.ha_url or "",
|
||||
token=settings.ha_token or "",
|
||||
)
|
||||
)
|
||||
app.state.ha_reader = HaReader(client=client)
|
||||
async def lifespan(app: FastAPI):
|
||||
settings = HaClientSettings(
|
||||
url=app.state.settings.ha_url,
|
||||
token=app.state.settings.ha_token,
|
||||
)
|
||||
client = HaClient(settings=settings)
|
||||
app.state.ha_reader = HaReader(client=client)
|
||||
yield
|
||||
|
||||
|
||||
@@ -32,9 +25,20 @@ app = FastAPI(
|
||||
lifespan=lifespan,
|
||||
)
|
||||
|
||||
register_exception_handlers(app)
|
||||
|
||||
app.include_router(entities_router)
|
||||
class Settings:
|
||||
ha_url: str
|
||||
ha_token: str
|
||||
|
||||
|
||||
app.state.settings = Settings()
|
||||
|
||||
|
||||
def get_ha_reader() -> HaReader:
|
||||
return app.state.ha_reader
|
||||
|
||||
|
||||
app.include_router(entities_router, dependencies=[Depends(get_ha_reader)])
|
||||
|
||||
|
||||
@app.get("/health")
|
||||
@@ -44,4 +48,4 @@ def health() -> dict[str, str]:
|
||||
|
||||
@app.get("/")
|
||||
def root() -> dict[str, str]:
|
||||
return {"service": "sillyhome-next", "docs": "/docs"}
|
||||
return {"service": "sillyhome-next", "docs": "/docs"}
|
||||
@@ -6,26 +6,10 @@ from app.ha.models import HaEntitySummary
|
||||
from app.rules.recommender import Rule
|
||||
|
||||
|
||||
HEATING_SENSOR_DEVICE_CLASSES = frozenset({"temperature", "humidity"})
|
||||
HEATING_BINARY_SENSOR_DEVICE_CLASSES = frozenset({"occupancy", "presence"})
|
||||
|
||||
|
||||
class HeatingRule(Rule):
|
||||
def matches(self, entities: Sequence[HaEntitySummary]) -> bool:
|
||||
for entity in entities:
|
||||
if entity.domain == "climate":
|
||||
return True
|
||||
if (
|
||||
entity.domain == "sensor"
|
||||
and entity.device_class in HEATING_SENSOR_DEVICE_CLASSES
|
||||
):
|
||||
return True
|
||||
if (
|
||||
entity.domain == "binary_sensor"
|
||||
and entity.device_class in HEATING_BINARY_SENSOR_DEVICE_CLASSES
|
||||
):
|
||||
return True
|
||||
return False
|
||||
domains = {item.domain for item in entities}
|
||||
return "climate" in domains or "sensor" in domains
|
||||
|
||||
def recommendation(self, entities: Sequence[HaEntitySummary]) -> str:
|
||||
return "Prüfe Heizungsregelung: Aktiviere energiesparenden Modus bei Abwesenheit."
|
||||
return "Prüfe Heizungsregelung: Aktiviere energiesparenden Modus bei Abwesenheit."
|
||||
@@ -7,12 +7,10 @@ dependencies = [
|
||||
"fastapi>=0.110.0",
|
||||
"uvicorn[standard]>=0.29.0",
|
||||
"pydantic>=2.6.0",
|
||||
"requests>=2.31.0",
|
||||
]
|
||||
|
||||
[project.optional-dependencies]
|
||||
dev = [
|
||||
"httpx2>=2.3.0",
|
||||
"pytest>=8.0.0",
|
||||
"ruff>=0.4.0",
|
||||
"mypy>=1.9.0",
|
||||
@@ -27,4 +25,4 @@ strict = true
|
||||
|
||||
[tool.ruff]
|
||||
line-length = 100
|
||||
target-version = "py311"
|
||||
target-version = "py311"
|
||||
@@ -1,4 +1,5 @@
|
||||
import requests
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
p = Path('/root/.openclaw/secrets/gitea.env')
|
||||
|
||||
@@ -1,63 +1,10 @@
|
||||
from collections.abc import Sequence
|
||||
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from app.ha.exceptions import HaTimeoutError
|
||||
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")]
|
||||
|
||||
|
||||
class TimeoutHaReader(HaReader):
|
||||
def __init__(self) -> None:
|
||||
pass
|
||||
|
||||
def read_entities(self) -> Sequence[HaEntitySummary]:
|
||||
raise HaTimeoutError("contains internal details that must not leak")
|
||||
client = TestClient(app)
|
||||
|
||||
|
||||
def test_openapi_docs_are_available() -> None:
|
||||
with TestClient(app) as client:
|
||||
response = client.get("/docs")
|
||||
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_class": None,
|
||||
"device_class": None,
|
||||
"unit_of_measurement": None,
|
||||
}
|
||||
]
|
||||
|
||||
|
||||
def test_entities_returns_503_without_home_assistant_config() -> None:
|
||||
with TestClient(app) as client:
|
||||
if hasattr(app.state, "ha_reader"):
|
||||
delattr(app.state, "ha_reader")
|
||||
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."}
|
||||
|
||||
@@ -1,79 +0,0 @@
|
||||
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()
|
||||
@@ -1,6 +1,7 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from app.ha.client import HaClient, HaClientSettings
|
||||
from app.ha.models import HaEntitySummary, HaState
|
||||
from app.ha.reader import HaReader
|
||||
|
||||
|
||||
|
||||
@@ -5,16 +5,8 @@ from app.rules.heating import HeatingRule
|
||||
from app.rules.recommender import Recommender
|
||||
|
||||
|
||||
def _sensor(entity_id: str, device_class: str | None = None) -> HaEntitySummary:
|
||||
return HaEntitySummary(entity_id=entity_id, domain="sensor", device_class=device_class)
|
||||
|
||||
|
||||
def _binary_sensor(entity_id: str, device_class: str | None = None) -> HaEntitySummary:
|
||||
return HaEntitySummary(
|
||||
entity_id=entity_id,
|
||||
domain="binary_sensor",
|
||||
device_class=device_class,
|
||||
)
|
||||
def _sensor(entity_id: str) -> HaEntitySummary:
|
||||
return HaEntitySummary(entity_id=entity_id, domain="sensor")
|
||||
|
||||
|
||||
def _climate(entity_id: str) -> HaEntitySummary:
|
||||
@@ -24,25 +16,11 @@ def _climate(entity_id: str) -> HaEntitySummary:
|
||||
def test_heating_rule_triggers() -> None:
|
||||
rule = HeatingRule()
|
||||
assert rule.matches([_climate("climate.living_room")])
|
||||
assert rule.matches([_sensor("sensor.temperature_living", device_class="temperature")])
|
||||
assert rule.matches([_sensor("sensor.humidity_bath", device_class="humidity")])
|
||||
assert rule.matches([_binary_sensor("binary_sensor.occupancy_living", "occupancy")])
|
||||
assert rule.matches([_binary_sensor("binary_sensor.presence_entry", "presence")])
|
||||
|
||||
|
||||
def test_heating_rule_ignores_non_relevant_sensors() -> None:
|
||||
rule = HeatingRule()
|
||||
assert not rule.matches([_sensor("sensor.temperature_living")])
|
||||
assert not rule.matches([_sensor("sensor.power", device_class="power")])
|
||||
assert not rule.matches([_sensor("sensor.voltage", device_class="voltage")])
|
||||
assert not rule.matches([_sensor("sensor.door", device_class="door")])
|
||||
assert not rule.matches([_sensor("sensor.window", device_class="window")])
|
||||
assert not rule.matches([_sensor("sensor.light", device_class="illuminance")])
|
||||
assert not rule.matches([_binary_sensor("binary_sensor.window", device_class="window")])
|
||||
assert rule.matches([_sensor("sensor.temperature_living")])
|
||||
|
||||
|
||||
def test_recommender_uses_rule() -> None:
|
||||
recommender = Recommender(rules=[HeatingRule()])
|
||||
assert recommender.run([_climate("climate.living_room")]) == [
|
||||
"Prüfe Heizungsregelung: Aktiviere energiesparenden Modus bei Abwesenheit."
|
||||
]
|
||||
]
|
||||
@@ -1,10 +1,10 @@
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from app.main import app
|
||||
|
||||
client = TestClient(app)
|
||||
|
||||
|
||||
def test_health_returns_ok() -> None:
|
||||
with TestClient(app) as client:
|
||||
response = client.get("/health")
|
||||
response = client.get("/health")
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {"status": "ok"}
|
||||
|
||||
Reference in New Issue
Block a user