HMS MediaEngine Agent
0922cc1d68
- Struktur gemäß §8 (Eigentumsgrenzen), PLAN.md als normative Basis - Pflichtdokumente: STATUS.md, ERRORS.md, TEST_REPORT.md, CHANGELOG.md, ADRs - ADR-0001 Python 3.13-Pin, ADR-0002 GStreamer 1.28.6-Pin (Windows), ADR-0003 IPC TCP+MessagePack v1 - Kernpakete: hms_protocol, hms_domain, hms_parameter, hms_artnet, hms_adaptive, hms_capabilities, hms_plugin_sdk - Renderer-Spike: D3D11-Primärpfad + Dev-GL-Pfad (§36 Nr. 4-5) - Control Core: FastAPI REST + WebSocket (§36 Nr. 9) - Beispielplugins: Passthrough + Gaussian Blur (3 Adaptive-Quality- Varianten, HLSL/GLSL/GLES) - Tools: Art-Net-Emulator, Fixture-Generator (Master32/Layer64-CSV), Capability-Probe - JSON-Schemas: IPC, Plugin, Projekt, Cluster - 121 Unit-/Integrationstests grün, Ruff grün Gate 0 bleibt offen: Hardwaremessungen nur auf echter Windows-Referenz- hardware gültig (§29.7, §33).
106 lines
4.2 KiB
Python
106 lines
4.2 KiB
Python
"""Unit-Tests Parameter-Engine (PLAN.md §11)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
|
|
import pytest
|
|
from hms_parameter import ParameterEngine, layer_opacity_path, master_intensity_path
|
|
from hms_parameter.engine import ControlSource, MergeMode, RevisionConflict
|
|
|
|
|
|
@pytest.fixture()
|
|
def path() -> str:
|
|
return layer_opacity_path(str(uuid.uuid4()), str(uuid.uuid4()))
|
|
|
|
|
|
def test_invalid_path_rejected() -> None:
|
|
engine = ParameterEngine()
|
|
with pytest.raises(ValueError, match="invalid parameter path"):
|
|
engine.set_value("composition/not-a-uuid/layer/x/opacity", 1.0, ControlSource.WEB)
|
|
with pytest.raises(ValueError, match="invalid parameter path"):
|
|
engine.set_value("../escape", 1.0, ControlSource.WEB)
|
|
|
|
|
|
def test_nan_and_infinity_rejected(path: str) -> None:
|
|
engine = ParameterEngine()
|
|
with pytest.raises(ValueError, match="finite"):
|
|
engine.set_value(path, float("nan"), ControlSource.WEB)
|
|
with pytest.raises(ValueError, match="finite"):
|
|
engine.set_value(path, float("inf"), ControlSource.WEB)
|
|
|
|
|
|
def test_higher_priority_wins(path: str) -> None:
|
|
engine = ParameterEngine()
|
|
engine.set_value(path, 0.5, ControlSource.WEB)
|
|
assert engine.effective_value(path) == pytest.approx(0.5)
|
|
engine.set_value(path, 0.9, ControlSource.CONSOLE)
|
|
assert engine.effective_value(path) == pytest.approx(0.9) # Pult überstimmt Web
|
|
engine.set_value(path, 0.0, ControlSource.SAFETY)
|
|
assert engine.effective_value(path) == pytest.approx(0.0) # Blackout überstimmt alles
|
|
|
|
|
|
def test_lower_priority_cannot_displace_higher(path: str) -> None:
|
|
engine = ParameterEngine()
|
|
engine.set_value(path, 0.9, ControlSource.CONSOLE)
|
|
engine.set_value(path, 0.1, ControlSource.WEB) # niedrigere Priorität
|
|
assert engine.effective_value(path) == pytest.approx(0.9) # CONSOLE bleibt wirksam
|
|
assert engine.current_source(path) is ControlSource.CONSOLE
|
|
|
|
|
|
def test_release_falls_back_to_lower_priority(path: str) -> None:
|
|
engine = ParameterEngine()
|
|
engine.set_value(path, 0.2, ControlSource.WEB)
|
|
engine.set_value(path, 0.8, ControlSource.CONSOLE)
|
|
engine.release(path, ControlSource.CONSOLE)
|
|
assert engine.effective_value(path) == pytest.approx(0.2) # Web übernimmt wieder
|
|
engine.release(path, ControlSource.WEB)
|
|
assert engine.current_source(path) is None
|
|
|
|
|
|
def test_revision_increments_on_set_and_release(path: str) -> None:
|
|
engine = ParameterEngine()
|
|
r0 = engine.revision
|
|
r1 = engine.set_value(path, 0.5, ControlSource.WEB)
|
|
r2 = engine.set_value(path, 0.6, ControlSource.WEB)
|
|
r3 = engine.release(path, ControlSource.WEB)
|
|
assert (r1, r2, r3) == (r0 + 1, r0 + 2, r0 + 3)
|
|
|
|
|
|
def test_optimistic_locking_revision_conflict(path: str) -> None:
|
|
engine = ParameterEngine()
|
|
engine.set_value(path, 0.5, ControlSource.WEB)
|
|
with pytest.raises(RevisionConflict):
|
|
engine.set_value(path, 0.6, ControlSource.WEB, expected_revision=engine.revision + 5)
|
|
engine.set_value(path, 0.6, ControlSource.WEB, expected_revision=engine.revision)
|
|
|
|
|
|
def test_snapshot_is_atomic_and_readonly(path: str) -> None:
|
|
engine = ParameterEngine()
|
|
engine.set_value(path, 0.42, ControlSource.WEB)
|
|
snap = engine.snapshot()
|
|
assert snap.get(path) == pytest.approx(0.42)
|
|
assert snap.revision == engine.revision
|
|
values = snap.as_dict()
|
|
values[path] = 99.0 # Manipulation der Kopie darf Snapshot nicht ändern
|
|
assert snap.get(path) == pytest.approx(0.42)
|
|
# Änderung nach dem Snapshot erscheint nicht im alten Snapshot (§11.4)
|
|
engine.set_value(path, 0.9, ControlSource.WEB)
|
|
assert snap.get(path) == pytest.approx(0.42)
|
|
|
|
|
|
def test_defaults_in_snapshot_without_override() -> None:
|
|
engine = ParameterEngine()
|
|
master = master_intensity_path()
|
|
engine.set_default(master, 1.0)
|
|
snap = engine.snapshot()
|
|
assert snap.get(master) == pytest.approx(1.0)
|
|
assert engine.effective_value(master) == pytest.approx(1.0)
|
|
|
|
|
|
def test_htp_mode_takes_maximum(path: str) -> None:
|
|
engine = ParameterEngine(merge_mode=MergeMode.HTP)
|
|
engine.set_value(path, 0.9, ControlSource.WEB)
|
|
engine.set_value(path, 0.1, ControlSource.WEB)
|
|
assert engine.effective_value(path) == pytest.approx(0.9) # HTP: Maximum bleibt
|