58 lines
1.7 KiB
Python
58 lines
1.7 KiB
Python
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from arbitrade.risk.stop_conditions import StopConditionsGuard
|
|
|
|
|
|
def test_stop_conditions_guard_halts_on_source_latency_breach() -> None:
|
|
guard = StopConditionsGuard(max_source_latency_ms=50.0)
|
|
|
|
guard.observe_latency(source_latency_ms=75.0, apply_latency_ms=1.0)
|
|
|
|
assert guard.is_halted
|
|
assert guard.halted_reason == "source_latency_limit_breached"
|
|
|
|
|
|
def test_stop_conditions_guard_halts_on_apply_latency_breach() -> None:
|
|
guard = StopConditionsGuard(max_apply_latency_ms=2.0)
|
|
|
|
guard.observe_latency(source_latency_ms=None, apply_latency_ms=3.5)
|
|
|
|
assert guard.is_halted
|
|
assert guard.halted_reason == "apply_latency_limit_breached"
|
|
|
|
|
|
def test_stop_conditions_guard_halts_on_consecutive_failures() -> None:
|
|
guard = StopConditionsGuard(max_consecutive_failures=2)
|
|
|
|
guard.register_failure()
|
|
assert not guard.is_halted
|
|
|
|
guard.register_failure()
|
|
|
|
assert guard.is_halted
|
|
assert guard.halted_reason == "consecutive_failures_limit_breached"
|
|
|
|
|
|
def test_stop_conditions_guard_resets_failures_after_success() -> None:
|
|
guard = StopConditionsGuard(max_consecutive_failures=3)
|
|
|
|
guard.register_failure()
|
|
guard.register_success()
|
|
guard.register_failure()
|
|
|
|
assert guard.consecutive_failures == 1
|
|
assert not guard.is_halted
|
|
|
|
|
|
def test_stop_conditions_guard_rejects_invalid_configuration() -> None:
|
|
with pytest.raises(ValueError, match="max_source_latency_ms"):
|
|
StopConditionsGuard(max_source_latency_ms=0.0)
|
|
|
|
with pytest.raises(ValueError, match="max_apply_latency_ms"):
|
|
StopConditionsGuard(max_apply_latency_ms=-1.0)
|
|
|
|
with pytest.raises(ValueError, match="max_consecutive_failures"):
|
|
StopConditionsGuard(max_consecutive_failures=0)
|