Refactor Buffer class PV disconnection logic for clarity

slac_timing/bsa_buffer.py

@@ -81,7 +81,7 @@ def _configure(self) -> None:
     def release(self) -> None:
         self._require_reserved("release")
         self.pvs.free.put(1)
-        self._disconnect_pvs()
+        self._clear_ca_cache()
 
     # --- Acquisition ---
 

slac_timing/buffer.py

@@ -2,7 +2,9 @@
 from abc import ABC, abstractmethod
 from typing import Optional
 
+import epics.ca
 import numpy as np
+from epics.pv import _PVcache_
 from pydantic import BaseModel, ConfigDict, PrivateAttr
 
 from slac_timing.pvs import BufferPVs
@@ -67,48 +69,6 @@ def release(self) -> None:
         """Release the buffer back to the system."""
         ...
 
-    def _disconnect_pvs(self) -> None:
-        """Disconnect all PV connections held by this buffer."""
-        if self._pvs is not None:
-            self._pvs.disconnect()
-            self._pvs = None
-        self._clear_ca_cache()
-
-    def _clear_ca_cache(self) -> None:
-        """Remove caget-created channels for this buffer from the CA cache."""
-        import epics.ca
-        from epics.pv import _PVcache_
-
-        ctx = epics.ca.current_context()
-        if ctx is None:
-            return
-
-        suffix = f"HST{self.number}"
-
-        stale_pvids = [
-            pvid for pvid in list(_PVcache_)
-            if pvid[0].endswith(suffix)
-        ]
-        for pvid in stale_pvids:
-            pv_obj = _PVcache_.pop(pvid, None)
-            if pv_obj is not None:
-                try:
-                    pv_obj.disconnect()
-                except BaseException:
-                    pass
-
-        context_cache = epics.ca._cache.get(ctx)
-        if context_cache is None:
-            return
-        stale_names = [name for name in context_cache if name.endswith(suffix)]
-        for name in stale_names:
-            entry = context_cache.get(name)
-            if entry is not None and getattr(entry, "chid", None) is not None:
-                try:
-                    epics.ca.clear_channel(entry.chid)
-                except BaseException:
-                    context_cache.pop(name, None)
-
     def is_reserved(self) -> bool:
         return self.number is not None and self.number != 0
 
@@ -293,4 +253,34 @@ def _fetch_many(self, epics, pvs: list[str]) -> dict[str, Optional[np.ndarray]]:
                 results[pv] = data
         return results
 
+    def _clear_ca_cache(self) -> None:
+        """Remove caget-created channels for this buffer from the CA cache."""
+        ctx = epics.ca.current_context()
+        if ctx is None:
+            return
+
+        suffix = f"HST{self.number}"
+
+        def clear_pv_object_cache():
+            stale_pvids = [
+                pvid for pvid in list(_PVcache_)
+                if pvid[0].endswith(suffix)
+            ]
+            for pvid in stale_pvids:
+                pv_obj = _PVcache_.pop(pvid, None)
+                if pv_obj is not None:
+                    pv_obj.disconnect()
+
+        def clear_context_cache():
+            context_cache = epics.ca._cache.get(ctx)
+            if context_cache is None:
+                return
+            stale_names = [name for name in context_cache if name.endswith(suffix)]
+            for name in stale_names:
+                entry = context_cache.get(name)
+                if entry is not None and getattr(entry, "chid", None) is not None:
+                    epics.ca.clear_channel(entry.chid)
+                    context_cache.pop(name, None)
 
+        clear_pv_object_cache()
+        clear_context_cache()

slac_timing/event_definition.py

@@ -102,7 +102,7 @@ def _configure(self) -> None:
     def release(self) -> None:
         self._require_reserved("release")
         self.pvs.free.put(1)
-        self._disconnect_pvs()
+        self._clear_ca_cache()
 
     # --- Acquisition ---
 

tests/test_buffer.py

@@ -1,4 +1,5 @@
-from unittest.mock import patch
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
 
 import numpy as np
 import pytest
@@ -156,3 +157,76 @@ def test_retries_batch(self, buffer):
         with patch("epics.caget_many", side_effect=[short_batch, ok_batch]):
             result = buffer.get_many(["PV:A", "PV:B"], retries=2, retry_delay=0)
         np.testing.assert_array_equal(result["PV:A"], np.arange(5, dtype=float))
+
+
+class TestClearCaCache:
+    """Tests for _clear_ca_cache and its nested helpers."""
+
+    def test_noop_when_no_context(self, buffer):
+        with patch("slac_timing.buffer.epics.ca") as mock_ca, \
+             patch("slac_timing.buffer._PVcache_", {}):
+            mock_ca.current_context.return_value = None
+            buffer._clear_ca_cache()
+            mock_ca._cache.get.assert_not_called()
+
+    def test_clears_matching_pv_objects(self, buffer):
+        pv_obj = MagicMock()
+        pv_cache = {
+            ("SOME:PV:HST1",): pv_obj,
+            ("OTHER:PV:HST2",): MagicMock(),
+        }
+        with patch("slac_timing.buffer.epics.ca") as mock_ca, \
+             patch("slac_timing.buffer._PVcache_", pv_cache):
+            mock_ca.current_context.return_value = "ctx"
+            mock_ca._cache.get.return_value = None
+            buffer._clear_ca_cache()
+
+        assert ("SOME:PV:HST1",) not in pv_cache
+        assert ("OTHER:PV:HST2",) in pv_cache
+        pv_obj.disconnect.assert_called_once()
+
+    def test_disconnect_exception_propagates(self, buffer):
+        pv_obj = MagicMock()
+        pv_obj.disconnect.side_effect = RuntimeError("dead channel")
+        pv_cache = {("SOME:PV:HST1",): pv_obj}
+        with patch("slac_timing.buffer.epics.ca") as mock_ca, \
+             patch("slac_timing.buffer._PVcache_", pv_cache):
+            mock_ca.current_context.return_value = "ctx"
+            mock_ca._cache.get.return_value = None
+            with pytest.raises(RuntimeError, match="dead channel"):
+                buffer._clear_ca_cache()
+
+    def test_clears_matching_context_cache_entries(self, buffer):
+        entry = SimpleNamespace(chid=42)
+        context_cache = {"SOME:PV:HST1": entry, "OTHER:PV:HST2": SimpleNamespace(chid=99)}
+        with patch("slac_timing.buffer.epics.ca") as mock_ca, \
+             patch("slac_timing.buffer._PVcache_", {}):
+            mock_ca.current_context.return_value = "ctx"
+            mock_ca._cache.get.return_value = context_cache
+            buffer._clear_ca_cache()
+
+        mock_ca.clear_channel.assert_called_once_with(42)
+        assert "OTHER:PV:HST2" in context_cache
+
+    def test_clear_channel_failure_propagates(self, buffer):
+        entry = SimpleNamespace(chid=42)
+        context_cache = {"SOME:PV:HST1": entry}
+        with patch("slac_timing.buffer.epics.ca") as mock_ca, \
+             patch("slac_timing.buffer._PVcache_", {}):
+            mock_ca.current_context.return_value = "ctx"
+            mock_ca._cache.get.return_value = context_cache
+            mock_ca.clear_channel.side_effect = RuntimeError("stale chid")
+            with pytest.raises(RuntimeError, match="stale chid"):
+                buffer._clear_ca_cache()
+
+    def test_skips_entry_without_chid(self, buffer):
+        entry = SimpleNamespace(chid=None)
+        context_cache = {"SOME:PV:HST1": entry}
+        with patch("slac_timing.buffer.epics.ca") as mock_ca, \
+             patch("slac_timing.buffer._PVcache_", {}):
+            mock_ca.current_context.return_value = "ctx"
+            mock_ca._cache.get.return_value = context_cache
+            buffer._clear_ca_cache()
+
+        mock_ca.clear_channel.assert_not_called()
+        assert "SOME:PV:HST1" in context_cache