sillyhome-next-dev/tests/api/test_actuators.py

from __future__ import annotations

from datetime import datetime, timedelta
from pathlib import Path

from fastapi.testclient import TestClient

from app.actuators.lifecycle import ActuatorReconciliationService
from app.actuators.store import ActuatorStore
from app.behavior.engine import BehaviorEngine
from app.config import Settings
from app.ha.discovery import DiscoveredEntity
from app.ha.discovery import discover_entities
from app.ha.history import (
    EntityHistorySeries,
    LogbookEntry,
    NumericHistoryPoint,
    StateHistorySeries,
)
from app.ha.models import HaEntitySummary
from app.ha.reader import HaReader
from app.main import app
from app.ml.registry.model_registry import ModelRegistry


class FakeHaReader(HaReader):
    def __init__(self, entities: list[HaEntitySummary], history: dict[str, list[float]]) -> None:
        self._entities = entities
        self._history = history

    def read_entities(self) -> list[HaEntitySummary]:
        return list(self._entities)

    def discover(
        self,
        domains: set[str] | None = None,
        learnable: bool | None = None,
    ) -> list[DiscoveredEntity]:
        return discover_entities(self._entities, domains=domains, learnable=learnable)

    def read_history(
        self,
        entity_ids: list[str],
        start_time: datetime,
        end_time: datetime,
    ) -> list[EntityHistorySeries]:
        base = start_time
        return [
            EntityHistorySeries(
                entity_id=entity_id,
                points=[
                    NumericHistoryPoint(
                        timestamp=base + timedelta(hours=index),
                        value=value,
                    )
                    for index, value in enumerate(self._history.get(entity_id, []))
                ],
            )
            for entity_id in entity_ids
            if entity_id in self._history
        ]

    def read_state_history(
        self,
        entity_ids: list[str],
        start_time: datetime,
        end_time: datetime,
    ) -> list[StateHistorySeries]:
        return []

    def read_logbook(
        self,
        entity_id: str,
        start_time: datetime,
        end_time: datetime,
    ) -> list[LogbookEntry]:
        return []

    def call_service(
        self,
        domain: str,
        service: str,
        service_data: dict[str, object],
    ) -> list[object]:
        return []


def _install_service(tmp_path: Path) -> None:
    entities = [
        HaEntitySummary(
            entity_id="light.abstellkammer",
            domain="light",
            friendly_name="Abstellkammer Licht",
            area_name="Abstellkammer",
        ),
        HaEntitySummary(
            entity_id="sensor.abstellkammer_illuminance",
            domain="sensor",
            device_class="illuminance",
            state_class="measurement",
            unit_of_measurement="lx",
            friendly_name="Abstellkammer Helligkeit",
            area_name="Abstellkammer",
        ),
        HaEntitySummary(
            entity_id="binary_sensor.abstellkammer_motion",
            domain="binary_sensor",
            device_class="motion",
            friendly_name="Abstellkammer Bewegung",
            area_name="Abstellkammer",
        ),
    ]
    settings = Settings(
        ha_url="http://ha.local",
        ha_token="token",
        model_store=str(tmp_path / "models"),
        automation_store=str(tmp_path / "automations"),
        actuator_store=str(tmp_path / "actuators"),
        history_days=14,
        min_training_points=5,
        retrain_stale_hours=24,
        reconcile_interval_seconds=900,
    )
    app.state.registry = ModelRegistry(tmp_path / "models")
    app.state.actuator_store = ActuatorStore(tmp_path / "actuators")
    app.state.ha_reader = FakeHaReader(
        entities,
        {"sensor.abstellkammer_illuminance": [10, 11, 12, 13, 14, 15]},
    )
    app.state.actuator_service = ActuatorReconciliationService(
        ha_reader=app.state.ha_reader,
        store=app.state.actuator_store,
        registry=app.state.registry,
        settings=settings,
    )
    app.state.behavior_engine = BehaviorEngine(
        ha_reader=app.state.ha_reader,
        store=app.state.actuator_store,
        settings=settings,
    )


def test_actuator_api_configures_reconciles_and_removes(tmp_path: Path) -> None:
    with TestClient(app) as client:
        _install_service(tmp_path)

        created = client.post("/v1/actuators", json={"actuator_entity_id": "light.abstellkammer"})
        assert created.status_code == 201
        assert created.json()["assignment"]["selected_numeric_entity_id"] == (
            "sensor.abstellkammer_illuminance"
        )

        listed = client.get("/v1/actuators")
        assert listed.status_code == 200
        assert listed.json()[0]["lifecycle"]["status"] == "trained"
        assert listed.json()[0]["behavior"]["mode"] == "shadow"

        evaluation = client.post("/v1/actuators/light.abstellkammer/evaluate")
        assert evaluation.status_code == 200

        premature_activation = client.post(
            "/v1/actuators/light.abstellkammer/activation",
            json={"active": True},
        )
        assert premature_activation.status_code == 409

        reconciliation = client.post("/v1/actuators/reconciliation/run")
        assert reconciliation.status_code == 200
        assert reconciliation.json()["trained_models"] == 1

        removed = client.delete("/v1/actuators/light.abstellkammer")
        assert removed.status_code == 204
        assert client.get("/v1/actuators").json() == []


def test_manual_override_endpoint_is_not_exposed(tmp_path: Path) -> None:
    with TestClient(app) as client:
        _install_service(tmp_path)
        client.post("/v1/actuators", json={"actuator_entity_id": "light.abstellkammer"})

        response = client.post(
            "/v1/actuators/light.abstellkammer/override",
            json={"numeric_entity_id": "sensor.abstellkammer_illuminance"},
        )

    assert response.status_code == 404