Merge pull request #78 from shilorigins/devagr/cl-return-com-error

MNT: return error rather than raising in ControlLayer._get_single

docs/source/upcoming_release_notes/78-cl_get_return_com_error.rst

@@ -0,0 +1,22 @@
+78 cl get return com error
+#################
+
+API Breaks
+----------
+- N/A
+
+Features
+--------
+- N/A
+
+Bugfixes
+--------
+- N/A
+
+Maintenance
+-----------
+- ControlLayer._get_single: return CommunicationError rather than propagating
+
+Contributors
+------------
+- shilorigins

superscore/control_layers/core.py

@@ -10,6 +10,7 @@
 
 from superscore.control_layers._base_shim import EpicsData
 from superscore.control_layers.status import TaskStatus
+from superscore.errors import CommunicationError
 
 from ._aioca import AiocaShim
 from ._base_shim import _BaseShim
@@ -96,12 +97,15 @@ def get(self, address: Union[str, Iterable[str]]) -> Union[EpicsData, Iterable[E
               "a string or list of strings")
 
     @get.register
-    def _get_single(self, address: str) -> EpicsData:
+    def _get_single(self, address: str) -> Union[EpicsData, CommunicationError]:
         """Synchronously get a single ``address``"""
-        return asyncio.run(self._get_one(address))
+        try:
+            return asyncio.run(self._get_one(address))
+        except CommunicationError as e:
+            return e
 
     @get.register
-    def _get_list(self, address: Iterable) -> Iterable[EpicsData]:
+    def _get_list(self, address: Iterable) -> Iterable[Union[EpicsData, CommunicationError]]:
         """Synchronously get a list of ``address``"""
         async def gathered_coros():
             coros = []

superscore/tests/test_cl.py

@@ -3,6 +3,7 @@
 import pytest
 
 from superscore.control_layers.status import TaskStatus
+from superscore.errors import CommunicationError
 
 
 def test_get(dummy_cl):
@@ -17,6 +18,12 @@ def test_get(dummy_cl):
     assert dummy_cl.get(['a', 'b', 'c']) == ["ca_value" for i in range(3)]
 
 
+def test_get_communication_error(dummy_cl):
+    mock_get = AsyncMock(side_effect=CommunicationError("Example error"))
+    dummy_cl.shims['ca'].get = mock_get
+    assert isinstance(dummy_cl.get("SOME_PREFIX"), CommunicationError)
+
+
 def test_put(dummy_cl):
     result = dummy_cl.put("OTHER:PREFIX", 4)
     assert isinstance(result, TaskStatus)