Support None or int with datetime index in ring buffer

Author: cwasicki

src/frequenz/sdk/timeseries/_ringbuffer/buffer.py

@@ -344,10 +344,12 @@ def window(
         if self.count_covered() == 0:
             return np.array([]) if isinstance(self._buffer, np.ndarray) else []
 
-        # If both are indices or None convert to datetime
-        if not isinstance(start, datetime) and not isinstance(end, datetime):
-            start, end = self._to_covered_indices(start, end)
+        # If there is an int index or None convert to datetime
+        if not isinstance(start, datetime):
+            start, _ = self._to_covered_indices(start, -1)
             start = self.get_timestamp(start)
+        if not isinstance(end, datetime):
+            _, end = self._to_covered_indices(0, end)
             end = self.get_timestamp(end)
 
         # Here we should have both as datetime

tests/timeseries/test_ringbuffer.py

@@ -525,12 +525,11 @@ def test_window_datetime() -> None:
     # Test datetime and Nones
     win = buffer.window(None, None)
     assert [0, np.nan, 2, 3, 4] == list(win)
-    with pytest.raises(ValueError):
-        win = buffer.window(dt(0), None)
-        #assert [0, np.nan, 2, 3, 4] == list(win)  # should not raise
-    with pytest.raises(ValueError):
-        win = buffer.window(None, dt(4))
-        #assert [0, np.nan, 2, 3, 4] == list(win)  # should not raise
+    win = buffer.window(dt(0), None)
+    assert [0, np.nan, 2, 3, 4] == list(win)
+    win = buffer.window(None, dt(5))
+    assert [0, np.nan, 2, 3, 4] == list(win)
+
     win = buffer.window(dt(0), dt(3), force_copy=False, fill_value=None)
     assert [0, np.nan, 2] == list(win)
     buffer._buffer[1] = 1  # pylint: disable=protected-access
@@ -606,15 +605,6 @@ def test_window_index_fill_value() -> None:
 def test_window_fail() -> None:
     """Test the window function with invalid arguments."""
     buffer = get_orb([0.0, 1.0, 2.0, 3.0, 4.0])
-    # Go crazy with the indices
-    with pytest.raises(IndexError):
-        buffer.window(dt(1), 3)
-    with pytest.raises(IndexError):
-        buffer.window(1, dt(3))
-    with pytest.raises(IndexError):
-        buffer.window(None, dt(2))
-    with pytest.raises(IndexError):
-        buffer.window(dt(2), None)
     # Invalid argument combination
     with pytest.raises(ValueError):
         buffer.window(0, 1, force_copy=False, fill_value=0)