155 lines
6.0 KiB
Python
155 lines
6.0 KiB
Python
from __future__ import annotations
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
import app.main as main
|
|
from app.core.event_core import EventCore
|
|
from app.core.models import EntityState, RuntimeState
|
|
from app.core.stores import FutureStores
|
|
|
|
|
|
def test_health_and_backup_roundtrip(tmp_path, monkeypatch) -> None: # type: ignore[no-untyped-def]
|
|
test_stores = FutureStores(tmp_path)
|
|
monkeypatch.setattr(main, "stores", test_stores)
|
|
monkeypatch.setattr(main, "event_core", EventCore(test_stores))
|
|
client = TestClient(main.app)
|
|
|
|
health = client.get("/health")
|
|
backup = client.get("/v2/backup/export")
|
|
restore = client.post("/v2/backup/restore", json=backup.json())
|
|
|
|
assert health.status_code == 200
|
|
assert health.json()["version"] == "2.0.0-alpha.12"
|
|
assert backup.status_code == 200
|
|
assert restore.status_code == 200
|
|
assert restore.json() == {"status": "restored"}
|
|
|
|
|
|
def test_dashboard_control_and_learning_endpoints(tmp_path, monkeypatch) -> None: # type: ignore[no-untyped-def]
|
|
test_stores = FutureStores(tmp_path)
|
|
test_stores.save_runtime(
|
|
RuntimeState(
|
|
entities={
|
|
"light.storage": EntityState(
|
|
entity_id="light.storage",
|
|
domain="light",
|
|
state="off",
|
|
),
|
|
"binary_sensor.storage_door": EntityState(
|
|
entity_id="binary_sensor.storage_door",
|
|
domain="binary_sensor",
|
|
state="off",
|
|
),
|
|
}
|
|
)
|
|
)
|
|
monkeypatch.setattr(main, "stores", test_stores)
|
|
monkeypatch.setattr(main, "event_core", EventCore(test_stores))
|
|
client = TestClient(main.app)
|
|
|
|
entities = client.get("/v2/entities?domain=light,binary_sensor")
|
|
control = client.post(
|
|
"/v2/control/light.storage/stage",
|
|
json={
|
|
"stage": "dry_run",
|
|
"min_confidence": 0.75,
|
|
"manual_block": False,
|
|
"cooldown_seconds": 120,
|
|
},
|
|
)
|
|
learning = client.post(
|
|
"/v2/learning/light.storage/patterns",
|
|
json={
|
|
"trigger_entity_id": "binary_sensor.storage_door",
|
|
"trigger_state": "on",
|
|
"target_state": "on",
|
|
"confidence": 0.91,
|
|
},
|
|
)
|
|
simulation = client.post(
|
|
"/v2/simulate",
|
|
json={
|
|
"actuator_entity_id": "light.storage",
|
|
"trigger_entity_id": "binary_sensor.storage_door",
|
|
"trigger_state": "on",
|
|
},
|
|
)
|
|
readiness = client.get("/v2/control/light.storage/readiness")
|
|
global_control = client.post("/v2/control/global", json={"enabled": False})
|
|
detailed_health = client.get("/v2/health")
|
|
feedback = client.post("/v2/control/light.storage/feedback", json={"kind": "wrong"})
|
|
summary = client.get("/v2/actuators/summary")
|
|
parity = client.get("/v2/feature-parity")
|
|
anomalies = client.get("/v2/anomalies")
|
|
proposal = client.post(
|
|
"/v2/automations/proposals",
|
|
json={
|
|
"name": "Storage light",
|
|
"trigger_entity_id": "binary_sensor.storage_door",
|
|
"trigger_state": "on",
|
|
"actuator_entity_id": "light.storage",
|
|
"target_state": "on",
|
|
},
|
|
)
|
|
proposal_yaml = client.get("/v2/automations/proposals/proposal-1/yaml")
|
|
approved = client.post("/v2/automations/proposals/proposal-1/approve")
|
|
weights = client.post(
|
|
"/v2/control/light.storage/weights",
|
|
json={"sensor_weights": {"binary_sensor.storage_door": 1.0}, "note": "door"},
|
|
)
|
|
history = client.post(
|
|
"/v2/history/analyze",
|
|
json={"entity_ids": ["binary_sensor.storage_door"]},
|
|
)
|
|
models = client.post("/v2/models/train", json={"actuator_entity_id": "light.storage"})
|
|
evaluation = client.post("/v2/models/evaluate", json={"actuator_entity_id": "light.storage"})
|
|
evaluations = client.get("/v2/models/evaluations")
|
|
jobs = client.get("/v2/jobs")
|
|
dashboard = client.get("/v2/dashboard")
|
|
|
|
assert entities.status_code == 200
|
|
assert [item["entity_id"] for item in entities.json()] == [
|
|
"binary_sensor.storage_door",
|
|
"light.storage",
|
|
]
|
|
assert control.status_code == 200
|
|
assert control.json()["stage"] == "dry_run"
|
|
assert learning.status_code == 200
|
|
assert learning.json()["patterns"][0]["trigger_entity_id"] == "binary_sensor.storage_door"
|
|
assert simulation.status_code == 200
|
|
assert simulation.json()["decision"]["target_state"] == "on"
|
|
assert readiness.status_code == 200
|
|
assert readiness.json()["ready"] is False
|
|
assert global_control.status_code == 200
|
|
assert global_control.json()["global_enabled"] is False
|
|
assert detailed_health.status_code == 200
|
|
assert detailed_health.json()["global_enabled"] is False
|
|
assert feedback.status_code == 200
|
|
assert feedback.json()["feedback_negative"] == 1
|
|
assert summary.status_code == 200
|
|
assert summary.json()[0]["actuator_entity_id"] == "light.storage"
|
|
assert parity.status_code == 200
|
|
assert "Job-Queue" in parity.json()["implemented"]
|
|
assert anomalies.status_code == 200
|
|
assert proposal.status_code == 200
|
|
assert proposal.json()["status"] == "draft"
|
|
assert proposal_yaml.status_code == 200
|
|
assert "alias: Storage light" in proposal_yaml.text
|
|
assert approved.status_code == 200
|
|
assert approved.json()["status"] == "approved"
|
|
assert weights.status_code == 200
|
|
assert weights.json()["sensor_weights"]["binary_sensor.storage_door"] == 1.0
|
|
assert history.status_code == 200
|
|
assert history.json()[0]["entity_id"] == "binary_sensor.storage_door"
|
|
assert models.status_code == 200
|
|
assert models.json()[0]["actuator_entity_id"] == "light.storage"
|
|
assert evaluation.status_code == 200
|
|
assert evaluation.json()[0]["verdict"] in {"bereit", "weiter testen"}
|
|
assert evaluations.status_code == 200
|
|
assert evaluations.json()
|
|
assert jobs.status_code == 200
|
|
assert jobs.json()
|
|
assert dashboard.status_code == 200
|
|
assert dashboard.json()["actuator_count"] == 1
|
|
assert dashboard.json()["automation_proposals"] == 1
|