Reject non-finite OOSM lag windows

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

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()