feat: Support restart and server-directed-shutdown-request in SSE contract test services

contract-tests/async_service.py

@@ -41,6 +41,8 @@ async def handle_get_status(request):
             'headers',
             'last-event-id',
             'read-timeout',
+            'restart',
+            'server-directed-shutdown-request'
         ]
     }
     return aiohttp.web.Response(

contract-tests/async_stream_entity.py

@@ -14,6 +14,7 @@
 from ld_eventsource.config.async_connect_strategy import \
     AsyncConnectStrategy  # noqa: E402
 from ld_eventsource.config.error_strategy import ErrorStrategy  # noqa: E402
+from ld_eventsource.errors import HTTPStatusError  # noqa: E402
 
 
 def millis_to_seconds(t):
@@ -26,6 +27,7 @@ def __init__(self, options, http_session: aiohttp.ClientSession):
         self.callback_url = options["callbackUrl"]
         self.log = logging.getLogger(options["tag"])
         self.closed = False
+        self._suppress_next_fault = False
         self.callback_counter = 0
         self.sse = None
         self._http_session = http_session
@@ -52,8 +54,11 @@ async def run(self):
                 initial_retry_delay=millis_to_seconds(self.options.get("initialDelayMs")),
                 last_event_id=self.options.get("lastEventId"),
                 error_strategy=ErrorStrategy.from_lambda(
-                    lambda _: (
-                        ErrorStrategy.FAIL if self.closed else ErrorStrategy.CONTINUE,
+                    lambda e: (
+                        ErrorStrategy.FAIL
+                        if self.closed
+                        or (isinstance(e, HTTPStatusError) and e.status == 204)
+                        else ErrorStrategy.CONTINUE,
                         None,
                     )
                 ),
@@ -80,6 +85,11 @@ async def run(self):
                     if self.closed:
                         break
                     if item.error:
+                        # See stream_entity.py: a caller-triggered interrupt surfaces as
+                        # an error on the read path; we don't report it to the harness.
+                        if self._suppress_next_fault:
+                            self._suppress_next_fault = False
+                            continue
                         self.log.info('Received error from stream: %s', item.error)
                         await self.send_message({'kind': 'error', 'error': str(item.error)})
         except Exception as e:
@@ -89,7 +99,11 @@ async def run(self):
 
     async def do_command(self, command: str) -> bool:
         self.log.info('Test service sent command: %s' % command)
-        # currently we support no special commands
+        if command == 'restart':
+            if self.sse:
+                self._suppress_next_fault = True
+                await self.sse.interrupt()
+            return True
         return False
 
     async def send_message(self, message):

contract-tests/service.py

@@ -48,7 +48,9 @@ def status():
             'comments',
             'headers',
             'last-event-id',
-            'read-timeout'
+            'read-timeout',
+            'restart',
+            'server-directed-shutdown-request'
         ]
     }
     return (json.dumps(body), 200, {'Content-type': 'application/json'})

contract-tests/stream_entity.py

@@ -12,6 +12,7 @@
 from ld_eventsource import *  # noqa: E402
 from ld_eventsource.actions import *  # noqa: E402
 from ld_eventsource.config import *  # noqa: E402
+from ld_eventsource.errors import HTTPStatusError  # noqa: E402
 
 http_client = urllib3.PoolManager()
 
@@ -26,6 +27,7 @@ def __init__(self, options):
         self.callback_url = options["callbackUrl"]
         self.log = logging.getLogger(options["tag"])
         self.closed = False
+        self._suppress_next_fault = False
         self.callback_counter = 0
         self.sse = None
 
@@ -56,8 +58,11 @@ def run(self):
                 ),
                 last_event_id=self.options.get("lastEventId"),
                 error_strategy=ErrorStrategy.from_lambda(
-                    lambda _: (
-                        ErrorStrategy.FAIL if self.closed else ErrorStrategy.CONTINUE,
+                    lambda e: (
+                        ErrorStrategy.FAIL
+                        if self.closed
+                        or (isinstance(e, HTTPStatusError) and e.status == 204)
+                        else ErrorStrategy.CONTINUE,
                         None,
                     )
                 ),
@@ -86,6 +91,12 @@ def run(self):
                     # item.error will be None if this is just an EOF rather than an I/O error or HTTP error.
                     # Currently the test harness does not expect us to send an error message in that case.
                     if item.error:
+                        # A caller-triggered interrupt ("restart" command) surfaces as an
+                        # error on the read path; like other services (e.g. okhttp's
+                        # StreamClosedByCallerException suppression) we don't report it.
+                        if self._suppress_next_fault:
+                            self._suppress_next_fault = False
+                            continue
                         self.log.info('Received error from stream: %s' % item.error)
                         self.send_message({'kind': 'error', 'error': str(item.error)})
         except Exception as e:
@@ -95,7 +106,11 @@ def run(self):
 
     def do_command(self, command: str) -> bool:
         self.log.info('Test service sent command: %s' % command)
-        # currently we support no special commands
+        if command == 'restart':
+            if self.sse:
+                self._suppress_next_fault = True
+                self.sse.interrupt()
+            return True
         return False
 
     def send_message(self, message):