80 lines
2.6 KiB
Python
80 lines
2.6 KiB
Python
|
"""
|
||
|
|
Tests for the Safety Framework.
|
||
|
|
"""
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
from src.core.safety import SafetyLimitEnforcer, SafetyLimits
|
||
|
|
|
||
|
|
|
||
|
|
class TestSafetyLimitEnforcer:
|
||
|
|
"""Test safety limit enforcement."""
|
||
|
|
|
||
|
|
def test_enforce_setpoint_within_limits(self):
|
||
|
|
"""Test setpoint within limits is not modified."""
|
||
|
|
enforcer = SafetyLimitEnforcer(None)
|
||
|
|
|
||
|
|
# Set up safety limits
|
||
|
|
limits = SafetyLimits(
|
||
|
|
hard_min_speed_hz=20.0,
|
||
|
|
hard_max_speed_hz=50.0,
|
||
|
|
hard_min_level_m=None,
|
||
|
|
hard_max_level_m=None,
|
||
|
|
hard_max_power_kw=None,
|
||
|
|
max_speed_change_hz_per_min=5.0
|
||
|
|
)
|
||
|
|
enforcer.safety_limits_cache[('station1', 'pump1')] = limits
|
||
|
|
|
||
|
|
# Test setpoint within limits
|
||
|
|
enforced, violations = enforcer.enforce_setpoint('station1', 'pump1', 35.0)
|
||
|
|
|
||
|
|
assert enforced == 35.0
|
||
|
|
assert violations == []
|
||
|
|
|
||
|
|
def test_enforce_setpoint_below_min(self):
|
||
|
|
"""Test setpoint below minimum is clamped."""
|
||
|
|
enforcer = SafetyLimitEnforcer(None)
|
||
|
|
|
||
|
|
limits = SafetyLimits(
|
||
|
|
hard_min_speed_hz=20.0,
|
||
|
|
hard_max_speed_hz=50.0,
|
||
|
|
hard_min_level_m=None,
|
||
|
|
hard_max_level_m=None,
|
||
|
|
hard_max_power_kw=None,
|
||
|
|
max_speed_change_hz_per_min=5.0
|
||
|
|
)
|
||
|
|
enforcer.safety_limits_cache[('station1', 'pump1')] = limits
|
||
|
|
|
||
|
|
enforced, violations = enforcer.enforce_setpoint('station1', 'pump1', 15.0)
|
||
|
|
|
||
|
|
assert enforced == 20.0
|
||
|
|
assert len(violations) == 1
|
||
|
|
assert "BELOW_MIN_SPEED" in violations[0]
|
||
|
|
|
||
|
|
def test_enforce_setpoint_above_max(self):
|
||
|
|
"""Test setpoint above maximum is clamped."""
|
||
|
|
enforcer = SafetyLimitEnforcer(None)
|
||
|
|
|
||
|
|
limits = SafetyLimits(
|
||
|
|
hard_min_speed_hz=20.0,
|
||
|
|
hard_max_speed_hz=50.0,
|
||
|
|
hard_min_level_m=None,
|
||
|
|
hard_max_level_m=None,
|
||
|
|
hard_max_power_kw=None,
|
||
|
|
max_speed_change_hz_per_min=5.0
|
||
|
|
)
|
||
|
|
enforcer.safety_limits_cache[('station1', 'pump1')] = limits
|
||
|
|
|
||
|
|
enforced, violations = enforcer.enforce_setpoint('station1', 'pump1', 55.0)
|
||
|
|
|
||
|
|
assert enforced == 50.0
|
||
|
|
assert len(violations) == 1
|
||
|
|
assert "ABOVE_MAX_SPEED" in violations[0]
|
||
|
|
|
||
|
|
def test_enforce_setpoint_no_limits(self):
|
||
|
|
"""Test setpoint without safety limits defined."""
|
||
|
|
enforcer = SafetyLimitEnforcer(None)
|
||
|
|
|
||
|
|
enforced, violations = enforcer.enforce_setpoint('station1', 'pump1', 35.0)
|
||
|
|
|
||
|
|
assert enforced == 0.0
|
||
|
|
assert violations == ["NO_SAFETY_LIMITS_DEFINED"]
|