42 lines
1.4 KiB
Python
42 lines
1.4 KiB
Python
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from app.ml.feature_store import FeatureStore, FeatureVector
|
|
|
|
|
|
def _vector(sensor_id: str, temperature: float, label: str | None = None) -> FeatureVector:
|
|
return FeatureVector(sensor_id=sensor_id, values={"temperature": temperature}, label=label)
|
|
|
|
|
|
def test_append_and_latest_returns_last_vector() -> None:
|
|
store = FeatureStore()
|
|
vectors = [_vector("sensor.living_room", 20.0), _vector("sensor.living_room", 21.5)]
|
|
for item in vectors:
|
|
store.add(item)
|
|
assert store.latest("sensor.living_room") == vectors[-1]
|
|
|
|
|
|
def test_latest_returns_none_when_empty() -> None:
|
|
store = FeatureStore()
|
|
assert store.latest("sensor.living_room") is None
|
|
|
|
|
|
def test_add_batch_appends_all_vectors() -> None:
|
|
store = FeatureStore()
|
|
vectors = [
|
|
_vector("sensor.kitchen", 19.0),
|
|
_vector("sensor.kitchen", 20.0),
|
|
_vector("sensor.bathroom", 23.5),
|
|
]
|
|
store.add_batch(vectors)
|
|
assert len(store.all()) == 3
|
|
assert store.latest("sensor.kitchen").values["temperature"] == 20.0
|
|
|
|
|
|
def test_different_sensors_are_stored_independently() -> None:
|
|
store = FeatureStore()
|
|
store.add(_vector("sensor.living_room", 21.0))
|
|
store.add(_vector("sensor.bedroom", 18.5))
|
|
assert store.latest("sensor.living_room").values["temperature"] == 21.0
|
|
assert store.latest("sensor.bedroom").values["temperature"] == 18.5 |