diff --git a/src/pyrecest/filters/out_of_sequence.py b/src/pyrecest/filters/out_of_sequence.py index 7bab68b96..b9076211e 100644 --- a/src/pyrecest/filters/out_of_sequence.py +++ b/src/pyrecest/filters/out_of_sequence.py @@ -35,6 +35,18 @@ def _coerce_time(time): return time +def _coerce_optional_max_lag(max_lag): + if max_lag is None: + return None + try: + max_lag = float(max_lag) + except (TypeError, ValueError, OverflowError) as exc: + raise ValueError("max_lag must be finite and nonnegative or None") from exc + if not isfinite(max_lag) or max_lag < 0.0: + raise ValueError("max_lag must be finite and nonnegative or None") + return max_lag + + def _coerce_bool_flag(value, name): if isinstance(value, bool): return value @@ -115,9 +127,7 @@ class FixedLagBuffer: """ def __init__(self, max_lag=None, maxlen=None, *, copy_values=True): - self.max_lag = None if max_lag is None else float(max_lag) - if self.max_lag is not None and self.max_lag < 0.0: - raise ValueError("max_lag must be nonnegative or None") + self.max_lag = _coerce_optional_max_lag(max_lag) self.maxlen = None if maxlen is None else int(maxlen) if self.maxlen is not None and self.maxlen <= 0: raise ValueError("maxlen must be positive or None") @@ -311,9 +321,7 @@ def _setup_replay(self, filter_object, initial_time=0.0, max_lag=None): self._filter_object = filter_object self._initial_time = _coerce_time(initial_time) self._latest_time = self._initial_time - self.max_lag = None if max_lag is None else float(max_lag) - if self.max_lag is not None and self.max_lag < 0.0: - raise ValueError("max_lag must be nonnegative or None") + self.max_lag = _coerce_optional_max_lag(max_lag) self._initial_state = _safe_deepcopy(self._filter_object.filter_state) self._events = [] self._next_sequence = 0 diff --git a/tests/filters/test_out_of_sequence_max_lag_validation.py b/tests/filters/test_out_of_sequence_max_lag_validation.py new file mode 100644 index 000000000..e3ad4ec72 --- /dev/null +++ b/tests/filters/test_out_of_sequence_max_lag_validation.py @@ -0,0 +1,37 @@ +import unittest + +from pyrecest.filters import ( + FixedLagBuffer, + MeasurementTimeBuffer, + OutOfSequenceParticleUpdater, +) + + +class _DummyParticleFilter: + def __init__(self): + self.filter_state = {"updates": 0} + + +class OutOfSequenceMaxLagValidationTest(unittest.TestCase): + def test_rejects_nonfinite_max_lag_across_public_entry_points(self): + constructors = { + "fixed_lag_buffer": lambda max_lag: FixedLagBuffer(max_lag=max_lag), + "measurement_time_buffer": lambda max_lag: MeasurementTimeBuffer( + max_lag=max_lag + ), + "particle_updater": lambda max_lag: OutOfSequenceParticleUpdater( + _DummyParticleFilter(), max_lag=max_lag + ), + } + + for max_lag in (float("nan"), float("inf"), -float("inf")): + for name, constructor in constructors.items(): + with self.subTest(max_lag=max_lag, constructor=name): + with self.assertRaisesRegex( + ValueError, "finite and nonnegative" + ): + constructor(max_lag) + + +if __name__ == "__main__": + unittest.main()