forked from cleveragents/cleveragents-core
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
87 lines
2.0 KiB
Python
87 lines
2.0 KiB
Python
"""ASV benchmarks for WebSocket client."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
_SRC = str(Path(__file__).resolve().parents[1] / "src")
|
|
if _SRC not in sys.path:
|
|
sys.path.insert(0, _SRC)
|
|
|
|
import cleveragents # noqa: E402
|
|
|
|
importlib.reload(cleveragents)
|
|
|
|
from cleveragents.a2a.models import A2aEvent # noqa: E402
|
|
from cleveragents.client.ws_client import ( # noqa: E402
|
|
ConnectionState,
|
|
EventDeduplicator,
|
|
WebSocketClient,
|
|
_ws_backoff_delay,
|
|
)
|
|
|
|
|
|
class DeduplicatorSuite:
|
|
"""Benchmark EventDeduplicator throughput."""
|
|
|
|
timeout = 60
|
|
|
|
def setup(self) -> None:
|
|
self.dedup = EventDeduplicator(capacity=1000)
|
|
|
|
def time_is_duplicate_miss(self) -> None:
|
|
d = EventDeduplicator(capacity=1000)
|
|
for i in range(100):
|
|
d.is_duplicate(f"evt-{i}")
|
|
|
|
def time_is_duplicate_hit(self) -> None:
|
|
d = EventDeduplicator(capacity=1000)
|
|
d.is_duplicate("evt-x")
|
|
for _ in range(100):
|
|
d.is_duplicate("evt-x")
|
|
|
|
|
|
class EventProcessingSuite:
|
|
"""Benchmark event processing throughput."""
|
|
|
|
timeout = 60
|
|
|
|
def setup(self) -> None:
|
|
self.client = WebSocketClient()
|
|
self.client._state.connected = True
|
|
self.client._running = True
|
|
self.client.subscribe(lambda e: None)
|
|
|
|
def time_process_event(self) -> None:
|
|
c = WebSocketClient()
|
|
c._state.connected = True
|
|
c._running = True
|
|
c.subscribe(lambda e: None)
|
|
for i in range(100):
|
|
event = A2aEvent(event_id=f"bench-{i}", event_type="plan.status")
|
|
c.process_event(event)
|
|
|
|
|
|
class BackoffSuite:
|
|
"""Benchmark backoff computation."""
|
|
|
|
timeout = 60
|
|
|
|
def time_backoff_delay(self) -> None:
|
|
for i in range(10):
|
|
_ws_backoff_delay(i, 1.0, 60.0)
|
|
|
|
|
|
class ConnectionStateSuite:
|
|
"""Benchmark ConnectionState operations."""
|
|
|
|
timeout = 60
|
|
|
|
def time_create_and_reset(self) -> None:
|
|
s = ConnectionState()
|
|
s.connected = True
|
|
s.last_event_id = "ev1"
|
|
s.reset()
|