Fix TemporalBufferManager test and API issues

- Change type hints from list to Sequence for extractor parameters
- Add public get_required_timespan() method to BufferProtocol and implementations
- Refactor tests to use public API instead of accessing private fields
- Fix test data to use proper DataArray instead of scalar Variables
- Use sc.identical() for DataArray comparisons instead of ==

Original task: Fix issues in TemporalBufferManager tests - private field access and type hint complaints about passing extractors (list vs Sequence).

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <[email protected]>

Author: SimonHeybrock

src/ess/livedata/dashboard/temporal_buffer_manager.py

@@ -5,7 +5,7 @@
 from __future__ import annotations
 
 import logging
-from collections.abc import Hashable, Iterator, Mapping
+from collections.abc import Hashable, Iterator, Mapping, Sequence
 from dataclasses import dataclass, field
 from typing import Generic, TypeVar
 
@@ -24,7 +24,9 @@ class _BufferState:
     """Internal state for a managed buffer."""
 
     buffer: BufferProtocol[sc.DataArray]
-    extractors: list[UpdateExtractor] = field(default_factory=list)
+    extractors: list[UpdateExtractor] = field(
+        default_factory=list
+    )  # Stored as list internally
 
 
 class TemporalBufferManager(Mapping[K, BufferProtocol[sc.DataArray]], Generic[K]):
@@ -77,7 +79,7 @@ def get_buffered_data(self, key: K) -> sc.DataArray | None:
             return None
         return self._states[key].buffer.get()
 
-    def create_buffer(self, key: K, extractors: list[UpdateExtractor]) -> None:
+    def create_buffer(self, key: K, extractors: Sequence[UpdateExtractor]) -> None:
         """
         Create a buffer with appropriate type based on extractors.
 
@@ -86,7 +88,7 @@ def create_buffer(self, key: K, extractors: list[UpdateExtractor]) -> None:
         key:
             Key to identify this buffer.
         extractors:
-            List of extractors that will use this buffer.
+            Sequence of extractors that will use this buffer.
         """
         if key in self._states:
             raise ValueError(f"Buffer with key {key} already exists")
@@ -133,7 +135,7 @@ def add_extractor(self, key: K, extractor: UpdateExtractor) -> None:
         state.extractors.append(extractor)
         self._reconfigure_buffer_if_needed(key, state)
 
-    def set_extractors(self, key: K, extractors: list[UpdateExtractor]) -> None:
+    def set_extractors(self, key: K, extractors: Sequence[UpdateExtractor]) -> None:
         """
         Replace all extractors for an existing buffer.
 
@@ -151,7 +153,7 @@ def set_extractors(self, key: K, extractors: list[UpdateExtractor]) -> None:
         key:
             Key identifying the buffer to update.
         extractors:
-            New list of extractors that will use this buffer.
+            New sequence of extractors that will use this buffer.
         """
         state = self._states[key]
         state.extractors = list(extractors)
@@ -227,7 +229,7 @@ def delete_buffer(self, key: K) -> None:
             del self._states[key]
 
     def _create_buffer_for_extractors(
-        self, extractors: list[UpdateExtractor]
+        self, extractors: Sequence[UpdateExtractor]
     ) -> BufferProtocol[sc.DataArray]:
         """
         Create appropriate buffer type based on extractors.
@@ -238,7 +240,7 @@ def _create_buffer_for_extractors(
         Parameters
         ----------
         extractors:
-            List of extractors that will use the buffer.
+            Sequence of extractors that will use the buffer.
 
         Returns
         -------
@@ -258,7 +260,9 @@ def _create_buffer_for_extractors(
             return TemporalBuffer()
 
     def _update_buffer_requirements(
-        self, buffer: BufferProtocol[sc.DataArray], extractors: list[UpdateExtractor]
+        self,
+        buffer: BufferProtocol[sc.DataArray],
+        extractors: Sequence[UpdateExtractor],
     ) -> None:
         """
         Update buffer requirements based on extractors.
@@ -271,7 +275,7 @@ def _update_buffer_requirements(
         buffer:
             The buffer to update.
         extractors:
-            List of extractors to gather requirements from.
+            Sequence of extractors to gather requirements from.
         """
         # Compute maximum required timespan
         if extractors:

src/ess/livedata/dashboard/temporal_buffers.py

@@ -52,6 +52,17 @@ def set_required_timespan(self, seconds: float) -> None:
             Required timespan in seconds.
         """
 
+    @abstractmethod
+    def get_required_timespan(self) -> float:
+        """
+        Get the required timespan for the buffer.
+
+        Returns
+        -------
+        :
+            Required timespan in seconds.
+        """
+
     @abstractmethod
     def set_max_memory(self, max_bytes: int) -> None:
         """
@@ -93,6 +104,10 @@ def set_required_timespan(self, seconds: float) -> None:
         """Set required timespan (unused for SingleValueBuffer)."""
         self._required_timespan = seconds
 
+    def get_required_timespan(self) -> float:
+        """Get the required timespan."""
+        return self._required_timespan
+
     def set_max_memory(self, max_bytes: int) -> None:
         """Set max memory (unused for SingleValueBuffer)."""
         self._max_memory = max_bytes
@@ -343,6 +358,10 @@ def set_required_timespan(self, seconds: float) -> None:
         """Set the required timespan for the buffer."""
         self._required_timespan = seconds
 
+    def get_required_timespan(self) -> float:
+        """Get the required timespan for the buffer."""
+        return self._required_timespan
+
     def set_max_memory(self, max_bytes: int) -> None:
         """Set the maximum memory usage for the buffer."""
         self._max_memory = max_bytes

tests/dashboard/temporal_buffer_manager_test.py

@@ -26,7 +26,7 @@ def test_create_buffer_with_only_latest_extractors_uses_single_value_buffer(self
 
         manager.create_buffer('test', extractors)
 
-        assert isinstance(manager._states['test'].buffer, SingleValueBuffer)
+        assert isinstance(manager['test'], SingleValueBuffer)
 
     def test_create_buffer_with_mixed_extractors_uses_temporal_buffer(self):
         """
@@ -37,7 +37,7 @@ def test_create_buffer_with_mixed_extractors_uses_temporal_buffer(self):
 
         manager.create_buffer('test', extractors)
 
-        assert isinstance(manager._states['test'].buffer, TemporalBuffer)
+        assert isinstance(manager['test'], TemporalBuffer)
 
     def test_create_buffer_with_window_extractor_uses_temporal_buffer(self):
         """Test that TemporalBuffer is used with WindowAggregatingExtractor."""
@@ -46,7 +46,7 @@ def test_create_buffer_with_window_extractor_uses_temporal_buffer(self):
 
         manager.create_buffer('test', extractors)
 
-        assert isinstance(manager._states['test'].buffer, TemporalBuffer)
+        assert isinstance(manager['test'], TemporalBuffer)
 
     def test_create_buffer_with_no_extractors_uses_single_value_buffer(self):
         """
@@ -56,7 +56,7 @@ def test_create_buffer_with_no_extractors_uses_single_value_buffer(self):
 
         manager.create_buffer('test', [])
 
-        assert isinstance(manager._states['test'].buffer, SingleValueBuffer)
+        assert isinstance(manager['test'], SingleValueBuffer)
 
     def test_create_buffer_raises_error_for_duplicate_key(self):
         """Test that creating a buffer with existing key raises ValueError."""
@@ -72,18 +72,24 @@ def test_update_buffer_adds_data(self):
         """Test that update_buffer adds data to the buffer."""
         manager = TemporalBufferManager()
         extractors = [LatestValueExtractor()]
-        data = sc.scalar(42, unit='counts')
+        data = sc.DataArray(
+            sc.scalar(42, unit='counts'),
+            coords={'time': sc.scalar(1.0, unit='s')},
+        )
 
         manager.create_buffer('test', extractors)
         manager.update_buffer('test', data)
 
         result = manager.get_buffered_data('test')
-        assert result == data
+        assert sc.identical(result, data)
 
     def test_update_buffer_raises_error_for_missing_key(self):
         """Test that updating non-existent buffer raises KeyError."""
         manager = TemporalBufferManager()
-        data = sc.scalar(42, unit='counts')
+        data = sc.DataArray(
+            sc.scalar(42, unit='counts'),
+            coords={'time': sc.scalar(1.0, unit='s')},
+        )
 
         with pytest.raises(KeyError, match="No buffer found"):
             manager.update_buffer('test', data)
@@ -94,12 +100,12 @@ def test_add_extractor_keeps_same_buffer_type(self):
         extractors = [LatestValueExtractor()]
 
         manager.create_buffer('test', extractors)
-        original_buffer = manager._states['test'].buffer
+        original_buffer = manager['test']
 
         manager.add_extractor('test', LatestValueExtractor())
 
-        assert manager._states['test'].buffer is original_buffer
-        assert isinstance(manager._states['test'].buffer, SingleValueBuffer)
+        assert manager['test'] is original_buffer
+        assert isinstance(manager['test'], SingleValueBuffer)
 
     def test_add_extractor_switches_to_temporal_buffer(self):
         """Test that switching buffer types preserves existing data."""
@@ -213,9 +219,9 @@ def test_window_extractor_sets_timespan_on_buffer(self):
 
         manager.create_buffer('test', extractors)
 
-        buffer = manager._states['test'].buffer
+        buffer = manager['test']
         assert isinstance(buffer, TemporalBuffer)
-        assert buffer._required_timespan == window_duration
+        assert buffer.get_required_timespan() == window_duration
 
     def test_multiple_window_extractors_use_max_timespan(self):
         """Test that maximum timespan from multiple extractors is used."""
@@ -228,8 +234,8 @@ def test_multiple_window_extractors_use_max_timespan(self):
 
         manager.create_buffer('test', extractors)
 
-        buffer = manager._states['test'].buffer
-        assert buffer._required_timespan == 5.0
+        buffer = manager['test']
+        assert buffer.get_required_timespan() == 5.0
 
     def test_latest_extractor_does_not_set_timespan(self):
         """Test that LatestValueExtractor doesn't set a timespan."""
@@ -238,9 +244,9 @@ def test_latest_extractor_does_not_set_timespan(self):
 
         manager.create_buffer('test', extractors)
 
-        buffer = manager._states['test'].buffer
+        buffer = manager['test']
         assert isinstance(buffer, SingleValueBuffer)
-        assert buffer._required_timespan == 0.0
+        assert buffer.get_required_timespan() == 0.0
 
     def test_mixed_extractors_use_window_timespan(self):
         """Test that timespan is set when mixing Latest and Window extractors."""
@@ -252,25 +258,25 @@ def test_mixed_extractors_use_window_timespan(self):
 
         manager.create_buffer('test', extractors)
 
-        buffer = manager._states['test'].buffer
+        buffer = manager['test']
         assert isinstance(buffer, TemporalBuffer)
-        assert buffer._required_timespan == 4.0
+        assert buffer.get_required_timespan() == 4.0
 
     def test_adding_extractor_updates_timespan(self):
         """Test that adding an extractor updates the buffer's timespan."""
         manager = TemporalBufferManager()
         extractors = [WindowAggregatingExtractor(window_duration_seconds=2.0)]
 
         manager.create_buffer('test', extractors)
-        buffer = manager._states['test'].buffer
-        assert buffer._required_timespan == 2.0
+        buffer = manager['test']
+        assert buffer.get_required_timespan() == 2.0
 
         # Add extractor with larger timespan
         manager.add_extractor(
             'test', WindowAggregatingExtractor(window_duration_seconds=10.0)
         )
 
-        assert buffer._required_timespan == 10.0
+        assert buffer.get_required_timespan() == 10.0
 
     def test_full_history_extractor_infinite_timespan(self):
         """Test that FullHistoryExtractor sets infinite timespan."""
@@ -279,9 +285,9 @@ def test_full_history_extractor_infinite_timespan(self):
 
         manager.create_buffer('test', extractors)
 
-        buffer = manager._states['test'].buffer
+        buffer = manager['test']
         assert isinstance(buffer, TemporalBuffer)
-        assert buffer._required_timespan == float('inf')
+        assert buffer.get_required_timespan() == float('inf')
 
     def test_full_history_with_window_uses_infinite(self):
         """Test that mixing FullHistory with Window uses infinite timespan."""
@@ -293,10 +299,10 @@ def test_full_history_with_window_uses_infinite(self):
 
         manager.create_buffer('test', extractors)
 
-        buffer = manager._states['test'].buffer
+        buffer = manager['test']
         assert isinstance(buffer, TemporalBuffer)
         # max(5.0, inf) = inf
-        assert buffer._required_timespan == float('inf')
+        assert buffer.get_required_timespan() == float('inf')
 
 
 class TestTemporalBufferManagerWithRealData: