From 607b43dc77c1a9d171368ea9481d20b44c998b51 Mon Sep 17 00:00:00 2001 From: Ryan Lamb <4955475+kinyoklion@users.noreply.github.com> Date: Wed, 28 Oct 2026 22:56:10 +0000 Subject: [PATCH] feat: support restart and 204-halt in contract test services Two capabilities the client already supports were not usable by the SSE contract test suite, in both the sync and async test services: - server-directed-shutdown-request: the library halts on a 204 by default, but the services' error_strategy override continued on every error, defeating it. The services now FAIL on HTTPStatusError with status 204 and CONTINUE otherwise (as before). - restart: the services returned 400 for the harness's restart command despite the client having interrupt(). The command now calls interrupt(), and the error that surfaces on the read path as a result is suppressed (same precedent as okhttp's service suppressing StreamClosedByCallerException). Verified: full contract test suite passes against both services, including the newly unlocked "204 halts re-connection attempts" and "caller can trigger a restart" tests. Generated-by: termag --- contract-tests/async_service.py | 2 ++ contract-tests/async_stream_entity.py | 20 +++++++++++++++++--- contract-tests/service.py | 4 +++- contract-tests/stream_entity.py | 21 ++++++++++++++++++--- 4 files changed, 40 insertions(+), 7 deletions(-) diff --git a/contract-tests/async_service.py b/contract-tests/async_service.py index 0c5a4ed..72a677f 100644 --- a/contract-tests/async_service.py +++ b/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( diff --git a/contract-tests/async_stream_entity.py b/contract-tests/async_stream_entity.py index 6ba5a1d..1a2fd55 100644 --- a/contract-tests/async_stream_entity.py +++ b/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): diff --git a/contract-tests/service.py b/contract-tests/service.py index 3fc173d..63c4ebe 100644 --- a/contract-tests/service.py +++ b/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'}) diff --git a/contract-tests/stream_entity.py b/contract-tests/stream_entity.py index 01ab52e..61bc2f1 100644 --- a/contract-tests/stream_entity.py +++ b/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):