Merge pull request #172 from simonsobs/koopman/seq-cleanup-error-hand…

…ling

Add error handling to SMuRF stream shutdown

src/sorunlib/smurf.py

@@ -13,6 +13,8 @@
 
 import time
 
+from ocs.client_http import ControlClientError
+
 import sorunlib as run
 from sorunlib._internal import check_response, check_started
 
@@ -366,7 +368,18 @@ def stream(state, tag=None, subtype=None):
     else:
         print('Stopping SMuRF streams.')
         for smurf in run.CLIENTS['smurf']:
-            smurf.stream.stop()
+            try:
+                smurf.stream.stop()
+            # Handles case where agent becomes unreachable
+            except ControlClientError as e:
+                print(f"Failed to stop stream on {smurf}, removing from targets list.")
+                print(e)
+                clients_to_remove.append(smurf)
+
+        # Remove failed SMuRF clients
+        for client in clients_to_remove:
+            run.CLIENTS['smurf'].remove(client)
+        clients_to_remove = []
 
         for smurf in run.CLIENTS['smurf']:
             print(f'Waiting for stream from {smurf.instance_id} to stop.')

tests/test_smurf.py

@@ -5,8 +5,10 @@
 
 import pytest
 
+from ocs.client_http import ControlClientError
 from ocs.ocs_client import OCSReply
 
+import sorunlib as run
 from sorunlib import smurf
 from util import create_patch_clients
 
@@ -17,6 +19,30 @@
 patch_clients = create_patch_clients('satp', autouse=True)
 
 
+# Used to mock errors communicating with an OCS Client
+class ErrorProcess:
+    def __init__(self, *args, **kwargs):
+        pass
+
+    def start(self, **kwargs):
+        raise ControlClientError("no callee registered for procedure <observatory.controller.ops")
+
+    def wait(self, timeout=None):
+        raise ControlClientError("no callee registered for procedure <observatory.controller.ops")
+
+    def status(self):
+        raise ControlClientError("no callee registered for procedure <observatory.controller.ops")
+
+    def stop(self):
+        raise ControlClientError("no callee registered for procedure <observatory.controller.ops")
+
+
+class ErrorClient:
+    def __init__(self, instance_id, *args, **kwargs):
+        self.stream = ErrorProcess()
+        self.instance_id = instance_id
+
+
 def test_set_targets():
     smurf.set_targets(['smurf1'])
     assert len(smurf.run.CLIENTS['smurf']) == 1
@@ -160,3 +186,9 @@ def test_stream_single_failure(state):
             client.stream.start.assert_called_once()
         else:
             client.stream.stop.assert_called_once()
+
+
+def test_stream_agent_unavailable_on_stop():
+    # Replace 'smurf1' client with one that will error on stream.stop()
+    run.CLIENTS['smurf'][0] = ErrorClient('smurf1')
+    smurf.stream(state='off')