40 lines
1.3 KiB
Python
40 lines
1.3 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from app.ml.feature_store import FeatureVector
|
|
from app.ml.registry.model_registry import ModelRegistry
|
|
from app.ml.retraining import RetrainingService, retrain_model
|
|
|
|
|
|
def _vector(sensor_id: str) -> FeatureVector:
|
|
return FeatureVector(sensor_id=sensor_id, values={"temperature": 21.0})
|
|
|
|
|
|
def test_retraining_registers_new_artifact(tmp_path: Path) -> None:
|
|
registry = ModelRegistry(tmp_path)
|
|
|
|
result = retrain_model(registry, "home-model", [_vector("sensor.kitchen")])
|
|
|
|
assert result.replaced is False
|
|
assert registry.load_artifact("home-model") == result.artifact
|
|
|
|
|
|
def test_retraining_replaces_existing_artifact(tmp_path: Path) -> None:
|
|
registry = ModelRegistry(tmp_path)
|
|
service = RetrainingService(registry)
|
|
service.retrain("home-model", [_vector("sensor.kitchen")])
|
|
|
|
result = service.retrain("home-model", [_vector("sensor.bedroom")])
|
|
|
|
assert result.replaced is True
|
|
assert result.artifact.supported_sensors == ("sensor.bedroom",)
|
|
assert ModelRegistry(tmp_path).load_artifact("home-model") == result.artifact
|
|
|
|
|
|
def test_retraining_rejects_empty_training_data(tmp_path: Path) -> None:
|
|
with pytest.raises(ValueError, match="keine Trainingsdaten"):
|
|
retrain_model(ModelRegistry(tmp_path), "home-model", [])
|