freemo
8e9aa7af48
- Promote _request() to public method request() in ServerHttpClient to fix encapsulation violation across sync_client and remote_project - Fix WebSocket connect() to raise NotImplementedError with clear TODO documenting that real websockets transport is not yet implemented - Fix thread safety: protect last_event_id write under self._lock in ws_client.process_event() - Broaden exception handling in sync() to catch ServerTimeoutError and A2aNotAvailableError in addition to ServerConnectionError - Add has_next to PageResult in benchmark and test helpers - Add comments explaining Retry-After header logging-only behavior and why blocking time.sleep is acceptable in sync client - Update all test steps, robot helpers, and benchmarks to use the renamed public request() method and updated connect() behavior Refs: #335, #336, #337, #338
311 lines
10 KiB
Python
311 lines
10 KiB
Python
"""Step definitions for WebSocket updates client feature tests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from behave import given, then, when
|
|
from behave.runner import Context
|
|
|
|
from cleveragents.a2a.models import A2aEvent
|
|
from cleveragents.client.exceptions import ServerConnectionError
|
|
from cleveragents.client.ws_client import (
|
|
ConnectionState,
|
|
EventDeduplicator,
|
|
WebSocketClient,
|
|
_ws_backoff_delay,
|
|
)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Construction
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a WebSocketClient with default settings")
|
|
def step_ws_default(context: Context) -> None:
|
|
context.ws_client = WebSocketClient()
|
|
context.call_error = None
|
|
|
|
|
|
@given('a WebSocketClient with url "{url}" and token "{token}"')
|
|
def step_ws_custom(context: Context, url: str, token: str) -> None:
|
|
context.ws_client = WebSocketClient(ws_url=url, api_token=token)
|
|
context.call_error = None
|
|
|
|
|
|
@given("a WebSocketClient with default settings and max_reconnects {n:d}")
|
|
def step_ws_max_reconnects(context: Context, n: int) -> None:
|
|
context.ws_client = WebSocketClient(
|
|
max_reconnects=n, reconnect_base=0.001, reconnect_max=0.01
|
|
)
|
|
context.call_error = None
|
|
|
|
|
|
@then('the ws_url should be "{expected}"')
|
|
def step_check_ws_url(context: Context, expected: str) -> None:
|
|
assert context.ws_client.ws_url == expected
|
|
|
|
|
|
@then("the heartbeat interval should be {expected:g}")
|
|
def step_check_heartbeat(context: Context, expected: float) -> None:
|
|
assert context.ws_client.heartbeat_interval == expected
|
|
|
|
|
|
@then("the dedup capacity should be {expected:d}")
|
|
def step_check_dedup_cap(context: Context, expected: int) -> None:
|
|
assert context.ws_client.dedup_capacity == expected
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Connection lifecycle
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I attempt to connect the ws client")
|
|
def step_ws_attempt_connect(context: Context) -> None:
|
|
try:
|
|
context.ws_client.connect()
|
|
except NotImplementedError as exc:
|
|
context.call_error = exc
|
|
|
|
|
|
@given("the ws client state is set to connected")
|
|
def step_ws_set_connected(context: Context) -> None:
|
|
context.ws_client._state.connected = True
|
|
context.ws_client._running = True
|
|
|
|
|
|
@when("I disconnect the ws client")
|
|
def step_ws_disconnect(context: Context) -> None:
|
|
context.ws_client.disconnect()
|
|
|
|
|
|
@then("the ws client should be connected")
|
|
def step_ws_connected(context: Context) -> None:
|
|
assert context.ws_client.connected is True
|
|
|
|
|
|
@then("the ws client should not be connected")
|
|
def step_ws_not_connected(context: Context) -> None:
|
|
assert context.ws_client.connected is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Reconnection
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I simulate a reconnect")
|
|
def step_ws_reconnect(context: Context) -> None:
|
|
context.ws_client._state.connected = False
|
|
context.ws_client.reconnect()
|
|
|
|
|
|
@when("I exhaust reconnect attempts")
|
|
def step_ws_exhaust_reconnects(context: Context) -> None:
|
|
context.ws_client._state.connected = True
|
|
context.ws_client._running = True
|
|
try:
|
|
for _ in range(context.ws_client._max_reconnects + 1):
|
|
context.ws_client._state.connected = False
|
|
context.ws_client.reconnect()
|
|
except ServerConnectionError as exc:
|
|
context.call_error = exc
|
|
|
|
|
|
@then("the reconnect count should be {expected:d}")
|
|
def step_check_reconnect_count(context: Context, expected: int) -> None:
|
|
assert context.ws_client.reconnect_count == expected
|
|
|
|
|
|
@then("a ServerConnectionError should be raised from ws client")
|
|
def step_ws_conn_error(context: Context) -> None:
|
|
assert isinstance(context.call_error, ServerConnectionError)
|
|
|
|
|
|
@then("a NotImplementedError should be raised from ws client")
|
|
def step_ws_not_implemented_error(context: Context) -> None:
|
|
assert isinstance(context.call_error, NotImplementedError)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Event processing
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a WebSocketClient with a subscriber")
|
|
def step_ws_with_subscriber(context: Context) -> None:
|
|
context.ws_client = WebSocketClient()
|
|
context.ws_client._state.connected = True
|
|
context.ws_client._running = True
|
|
context.received_events: list[A2aEvent] = []
|
|
|
|
def _on_event(event: A2aEvent) -> None:
|
|
context.received_events.append(event)
|
|
|
|
context.ws_callback = _on_event
|
|
context.ws_client.subscribe(_on_event)
|
|
context.call_error = None
|
|
|
|
|
|
@when('I process a plan status event with id "{event_id}"')
|
|
def step_process_event(context: Context, event_id: str) -> None:
|
|
event = A2aEvent(
|
|
event_id=event_id,
|
|
event_type="plan.status",
|
|
plan_id="plan-001",
|
|
data={"status": "running"},
|
|
)
|
|
context.ws_process_result = context.ws_client.process_event(event)
|
|
|
|
|
|
@when('I process a plan status event with id "{event_id}" again')
|
|
def step_process_event_again(context: Context, event_id: str) -> None:
|
|
event = A2aEvent(
|
|
event_id=event_id,
|
|
event_type="plan.status",
|
|
plan_id="plan-001",
|
|
data={"status": "running"},
|
|
)
|
|
context.ws_process_result = context.ws_client.process_event(event)
|
|
|
|
|
|
@when("I unsubscribe the callback")
|
|
def step_unsubscribe(context: Context) -> None:
|
|
context.ws_client.unsubscribe(context.ws_callback)
|
|
|
|
|
|
@then('the subscriber should receive event "{event_id}"')
|
|
def step_check_received(context: Context, event_id: str) -> None:
|
|
assert any(e.event_id == event_id for e in context.received_events)
|
|
|
|
|
|
@then("the subscriber should have received {n:d} event")
|
|
def step_check_event_count_singular(context: Context, n: int) -> None:
|
|
assert len(context.received_events) == n
|
|
|
|
|
|
@then("the subscriber should have received {n:d} events")
|
|
def step_check_event_count_plural(context: Context, n: int) -> None:
|
|
assert len(context.received_events) == n
|
|
|
|
|
|
@then('the last event id should be "{expected}"')
|
|
def step_check_last_event_id(context: Context, expected: str) -> None:
|
|
assert context.ws_client.last_event_id == expected
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Heartbeat
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I handle a heartbeat")
|
|
def step_handle_heartbeat(context: Context) -> None:
|
|
context.ws_client.handle_heartbeat()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Version negotiation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when('I negotiate ws version "{version}"')
|
|
def step_negotiate_ws(context: Context, version: str) -> None:
|
|
context.ws_negotiated = context.ws_client.negotiate_version(version)
|
|
|
|
|
|
@then('the negotiated ws version should be "{expected}"')
|
|
def step_check_ws_version(context: Context, expected: str) -> None:
|
|
assert context.ws_negotiated == expected
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# EventDeduplicator
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("an EventDeduplicator with capacity {n:d}")
|
|
def step_dedup(context: Context, n: int) -> None:
|
|
context.dedup = EventDeduplicator(capacity=n)
|
|
context.dedup_result = False
|
|
|
|
|
|
@when('I check event "{event_id}" for duplicate')
|
|
def step_check_dedup(context: Context, event_id: str) -> None:
|
|
context.dedup_result = context.dedup.is_duplicate(event_id)
|
|
|
|
|
|
@then("the dedup result should be not duplicate")
|
|
def step_not_dup(context: Context) -> None:
|
|
assert context.dedup_result is False
|
|
|
|
|
|
@then("the dedup result should be duplicate")
|
|
def step_is_dup(context: Context) -> None:
|
|
assert context.dedup_result is True
|
|
|
|
|
|
@when('I add events "{a}" and "{b}" and "{c}" to the deduplicator')
|
|
def step_add_events(context: Context, a: str, b: str, c: str) -> None:
|
|
context.dedup.is_duplicate(a)
|
|
context.dedup.is_duplicate(b)
|
|
context.dedup.is_duplicate(c)
|
|
|
|
|
|
@then('event "{event_id}" should not be tracked')
|
|
def step_not_tracked(context: Context, event_id: str) -> None:
|
|
# If it's not tracked, calling is_duplicate should return False
|
|
# (it will also add it, but that's fine for the test)
|
|
assert event_id not in context.dedup._seen
|
|
|
|
|
|
@then("the deduplicator size should be {expected:d}")
|
|
def step_dedup_size(context: Context, expected: int) -> None:
|
|
assert context.dedup.size == expected
|
|
|
|
|
|
@when("I clear the deduplicator")
|
|
def step_clear_dedup(context: Context) -> None:
|
|
context.dedup.clear()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ConnectionState
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given('a ConnectionState with connected true and last_event_id "{eid}"')
|
|
def step_conn_state(context: Context, eid: str) -> None:
|
|
context.conn_state = ConnectionState()
|
|
context.conn_state.connected = True
|
|
context.conn_state.last_event_id = eid
|
|
|
|
|
|
@when("I reset the connection state")
|
|
def step_reset_state(context: Context) -> None:
|
|
context.conn_state.reset()
|
|
|
|
|
|
@then("the connection state should be disconnected")
|
|
def step_state_disconnected(context: Context) -> None:
|
|
assert context.conn_state.connected is False
|
|
|
|
|
|
@then("the connection state last_event_id should be empty")
|
|
def step_state_no_event_id(context: Context) -> None:
|
|
assert context.conn_state.last_event_id == ""
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Backoff
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I compute ws backoff delay for attempt {a:d} base {b:g} max {m:g}")
|
|
def step_ws_backoff(context: Context, a: int, b: float, m: float) -> None:
|
|
context.ws_delay = _ws_backoff_delay(a, b, m)
|
|
|
|
|
|
@then("the ws delay should be {expected:g}")
|
|
def step_check_ws_delay(context: Context, expected: float) -> None:
|
|
assert context.ws_delay == expected
|