From 471146761eb0053d1ff1886ae9933a0d9e00dc24 Mon Sep 17 00:00:00 2001 From: Otto Date: Thu, 11 Jun 2026 21:06:39 +0200 Subject: [PATCH] harden model registry persistence and evaluation --- app/ml/evaluation.py | 31 +++++++++++------- app/ml/registry/model_registry.py | 53 ++++++++++++++++++++++++++++--- tests/ml/test_evaluation.py | 26 +++++++++++++-- tests/ml/test_model_registry.py | 38 ++++++++++++++++++++++ 4 files changed, 129 insertions(+), 19 deletions(-) create mode 100644 tests/ml/test_model_registry.py diff --git a/app/ml/evaluation.py b/app/ml/evaluation.py index f0bcb2a..e163ebf 100644 --- a/app/ml/evaluation.py +++ b/app/ml/evaluation.py @@ -1,11 +1,10 @@ from __future__ import annotations import logging -from dataclasses import dataclass, field -from typing import Sequence +from collections.abc import Sequence +from dataclasses import dataclass -from app.ml.feature_store import FeatureVector -from app.ml.training import TrainingPipeline, TrainedArtifact +from app.ml.training import TrainingPipeline logger = logging.getLogger(__name__) @@ -29,15 +28,16 @@ class Evaluator: self._pipeline = pipeline def evaluate(self, artifact_id: str, predictions: Sequence[str]) -> EvalReport: - artifacts = list(self._pipeline._artifacts) - if not artifacts: - raise ValueError("Kein trainiertes Modell für Evaluation vorhanden.") + try: + supported_sensors = set(self._pipeline.export(artifact_id).supported_sensors) + except KeyError as exc: + raise ValueError("Kein trainiertes Modell für Evaluation vorhanden.") from exc - supported_sensors = self._pipeline.export(artifact_id).supported_sensors - unknown_hits = sum(1 for prediction in predictions if ":" not in prediction) - supported_references = sum(1 for sensor in supported_sensors for prediction in predictions if sensor in prediction) + parsed_sensors = [_prediction_sensor(prediction) for prediction in predictions] + supported_hits = sum(sensor in supported_sensors for sensor in parsed_sensors) + unknown_hits = sum(sensor not in supported_sensors for sensor in parsed_sensors) sample_size = len(predictions) - coverage = supported_references / sample_size if sample_size else 0.0 + coverage = supported_hits / sample_size if sample_size else 0.0 unknown_rate = unknown_hits / sample_size if sample_size else 0.0 coverage_metric = Metric(name="coverage", value=coverage, threshold=0.8) @@ -54,4 +54,11 @@ class Evaluator: coverage, unknown_rate, ) - return report \ No newline at end of file + return report + + +def _prediction_sensor(prediction: str) -> str | None: + parts = prediction.split(":", 2) + if len(parts) != 3 or not parts[0] or not parts[1]: + return None + return parts[1] diff --git a/app/ml/registry/model_registry.py b/app/ml/registry/model_registry.py index a236d24..94d2eac 100644 --- a/app/ml/registry/model_registry.py +++ b/app/ml/registry/model_registry.py @@ -1,26 +1,34 @@ from __future__ import annotations +import json import logging +import os from pathlib import Path -from typing import Iterable +import re +from collections.abc import Iterable from app.ml.training import TrainedArtifact logger = logging.getLogger(__name__) +_ARTIFACT_ID_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$") + class ModelRegistry: def __init__(self, root: str | Path) -> None: - self._root = Path(root) + self._root = Path(root).resolve() self._root.mkdir(parents=True, exist_ok=True) self._artifacts: dict[str, TrainedArtifact] = {} + self._load_existing() def register(self, artifact: TrainedArtifact) -> TrainedArtifact: + self._validate_artifact_id(artifact.artifact_id) self._artifacts[artifact.artifact_id] = artifact self._persist(artifact) return artifact def load_artifact(self, artifact_id: str) -> TrainedArtifact: + self._validate_artifact_id(artifact_id) if artifact_id not in self._artifacts: raise KeyError(f"Artifact '{artifact_id}' nicht registriert.") return self._artifacts[artifact_id] @@ -28,10 +36,45 @@ class ModelRegistry: def list_models(self) -> Iterable[TrainedArtifact]: return list(self._artifacts.values()) + def _load_existing(self) -> None: + for source in sorted(self._root.glob("*.json")): + try: + raw = json.loads(source.read_text(encoding="utf-8")) + artifact_id = raw["artifact_id"] + supported_sensors = raw["supported_sensors"] + if not isinstance(artifact_id, str) or not isinstance(supported_sensors, list): + raise ValueError("invalid artifact structure") + self._validate_artifact_id(artifact_id) + if source.name != f"{artifact_id}.json": + raise ValueError("artifact id does not match filename") + if not all(isinstance(sensor, str) for sensor in supported_sensors): + raise ValueError("supported_sensors must contain strings") + except (KeyError, TypeError, ValueError, json.JSONDecodeError) as exc: + raise ValueError(f"Ungültiges Modell-Artefakt: {source.name}") from exc + + self._artifacts[artifact_id] = TrainedArtifact( + artifact_id=artifact_id, + supported_sensors=tuple(supported_sensors), + ) + def _persist(self, artifact: TrainedArtifact) -> None: target = self._root / f"{artifact.artifact_id}.json" - target.write_text( - f"{artifact.artifact_id}\t{','.join(artifact.supported_sensors)}\n", + temporary = target.with_suffix(".json.tmp") + payload = { + "artifact_id": artifact.artifact_id, + "supported_sensors": list(artifact.supported_sensors), + } + temporary.write_text( + json.dumps(payload, ensure_ascii=True, sort_keys=True) + "\n", encoding="utf-8", ) - logger.info("Modell gespeichert: %s", target) \ No newline at end of file + os.replace(temporary, target) + logger.info("Modell gespeichert: %s", target) + + @staticmethod + def _validate_artifact_id(artifact_id: str) -> None: + if not _ARTIFACT_ID_PATTERN.fullmatch(artifact_id) or ".." in artifact_id: + raise ValueError( + "artifact_id darf nur Buchstaben, Ziffern, Punkt, Unterstrich " + "und Bindestrich enthalten." + ) diff --git a/tests/ml/test_evaluation.py b/tests/ml/test_evaluation.py index fac9894..d54182a 100644 --- a/tests/ml/test_evaluation.py +++ b/tests/ml/test_evaluation.py @@ -21,13 +21,35 @@ def evaluator_factory() -> Evaluator: def test_evaluate_returns_report_with_metrics() -> None: evaluator = evaluator_factory() - report = evaluator.evaluate("artifact_v1", ["artifact_v1:sensor.kitchen:{'temperature': 21.0}", "artifact_v1:sensor.bedroom:{'temperature': 18.5}"]) + report = evaluator.evaluate( + "artifact_v1", + [ + "artifact_v1:sensor.kitchen:{'temperature': 21.0}", + "artifact_v1:sensor.bedroom:{'temperature': 18.5}", + ], + ) assert report.artifact_id == "artifact_v1" assert report.sample_size == 2 assert {metric.name for metric in report.metrics} == {"coverage", "unknown_rate"} + assert next(metric.value for metric in report.metrics if metric.name == "coverage") == 1.0 def test_evaluate_without_training_raises_value_error() -> None: evaluator = Evaluator(TrainingPipeline(FeatureStore())) with pytest.raises(ValueError): - evaluator.evaluate("artifact_v1", []) \ No newline at end of file + evaluator.evaluate("artifact_v1", []) + + +def test_coverage_is_bounded_and_requires_exact_sensor_match() -> None: + evaluator = evaluator_factory() + report = evaluator.evaluate( + "artifact_v1", + [ + "artifact_v1:sensor.kitchen:{'note': 'sensor.bedroom'}", + "artifact_v1:sensor.kitchen_extra:{}", + "malformed", + ], + ) + + metrics = {metric.name: metric.value for metric in report.metrics} + assert metrics == {"coverage": pytest.approx(1 / 3), "unknown_rate": pytest.approx(2 / 3)} diff --git a/tests/ml/test_model_registry.py b/tests/ml/test_model_registry.py new file mode 100644 index 0000000..3558ae8 --- /dev/null +++ b/tests/ml/test_model_registry.py @@ -0,0 +1,38 @@ +from __future__ import annotations + +import json + +import pytest + +from app.ml.registry.model_registry import ModelRegistry +from app.ml.training import TrainedArtifact + + +def test_registry_loads_persisted_artifacts_after_restart(tmp_path) -> None: + registry = ModelRegistry(tmp_path) + artifact = TrainedArtifact("model-v1", ("sensor.kitchen", "sensor.bedroom")) + registry.register(artifact) + + restarted = ModelRegistry(tmp_path) + + assert restarted.load_artifact("model-v1") == artifact + + +@pytest.mark.parametrize("artifact_id", ["../escape", "nested/model", "..", ""]) +def test_registry_rejects_unsafe_artifact_ids(tmp_path, artifact_id: str) -> None: + registry = ModelRegistry(tmp_path) + + with pytest.raises(ValueError): + registry.register(TrainedArtifact(artifact_id, ("sensor.kitchen",))) + + assert list(tmp_path.parent.glob("escape.json")) == [] + + +def test_registry_rejects_corrupt_persisted_artifact(tmp_path) -> None: + (tmp_path / "broken.json").write_text( + json.dumps({"artifact_id": "../broken", "supported_sensors": []}), + encoding="utf-8", + ) + + with pytest.raises(ValueError, match="broken.json"): + ModelRegistry(tmp_path)