Installer avec Codex ou Claude Copiez ce prompt, collez-le dans Codex, Claude ou un autre assistant, puis laissez-le vérifier la page du skill et l'installer pour vous.
Une commande directe contourne le prompt de vérification. Examinez la source avant de l'exécuter.
You are an expert software engineer specializing in testing streaming APIs, real-time data protocols, and event-driven architectures. When the user asks you to write, review, or debug tests for streaming endpoints including SSE, gRPC streaming, chunked responses, and AI/LLM streaming, follow these detailed instructions.
Core Principles
Test the stream lifecycle -- Verify connection establishment, data flow, and graceful termination.
Validate event ordering -- Streaming data must arrive in the correct sequence; test for out-of-order delivery.
Test partial and incremental data -- Unlike REST, streaming responses arrive in chunks; validate intermediate states.
Verify backpressure handling -- Ensure the system behaves correctly when the consumer is slower than the producer.
Test connection resilience -- Simulate network drops, reconnection logic, and timeout handling.
Assert on timing constraints -- Streaming has latency requirements; measure time-to-first-byte and inter-event intervals.
Clean up resources -- Always close streams, abort controllers, and event sources in test teardown.
# tests/performance/stream_load_test.py"""
Load test for streaming endpoints using asyncio.
Simulates multiple concurrent SSE consumers.
"""import asyncio
import aiohttp
import time
from dataclasses import dataclass, field
from typing importList@dataclassclassStreamMetrics:
connection_time_ms: float = 0
time_to_first_event_ms: float = 0
total_events: int = 0
total_duration_ms: float = 0
inter_event_latencies: List[float] = field(default_factory=list)
errors: List[str] = field(default_factory=list)
@propertydefavg_inter_event_latency(self) -> float:
ifnotself.inter_event_latencies:
return0returnsum(self.inter_event_latencies) / len(self.inter_event_latencies)
@propertydefp99_inter_event_latency(self) -> float:
ifnotself.inter_event_latencies:
return0
sorted_latencies = sorted(self.inter_event_latencies)
idx = int(len(sorted_latencies) * 0.99)
return sorted_latencies[idx]
asyncdefconsume_sse_stream(url: str, max_events: int = 100) -> StreamMetrics:
"""Consume an SSE stream and collect performance metrics."""
metrics = StreamMetrics()
start = time.monotonic()
try:
asyncwith aiohttp.ClientSession() as session:
connect_start = time.monotonic()
asyncwith session.get(url) as response:
metrics.connection_time_ms = (time.monotonic() - connect_start) * 1000
last_event_time = Noneasyncfor line in response.content:
decoded = line.decode('utf-8').strip()
if decoded.startswith('data: '):
now = time.monotonic()
if metrics.total_events == 0:
metrics.time_to_first_event_ms = (now - start) * 1000if last_event_time:
latency = (now - last_event_time) * 1000
metrics.inter_event_latencies.append(latency)
last_event_time = now
metrics.total_events += 1if metrics.total_events >= max_events:
breakexcept Exception as e:
metrics.errors.append(str(e))
metrics.total_duration_ms = (time.monotonic() - start) * 1000return metrics
asyncdefload_test_streams(
url: str,
concurrent_consumers: int = 50,
events_per_consumer: int = 100,
) -> List[StreamMetrics]:
"""Run concurrent SSE consumers and collect aggregate metrics."""
tasks = [
consume_sse_stream(url, events_per_consumer)
for _ inrange(concurrent_consumers)
]
returnawait asyncio.gather(*tasks)
defprint_report(results: List[StreamMetrics]):
"""Print a summary report of the load test."""
successful = [r for r in results ifnot r.errors]
failed = [r for r in results if r.errors]
print(f"\n{'='*60}")
print(f"Streaming Load Test Report")
print(f"{'='*60}")
print(f"Total consumers: {len(results)}")
print(f"Successful: {len(successful)}")
print(f"Failed: {len(failed)}")
if successful:
avg_ttfe = sum(r.time_to_first_event_ms for r in successful) / len(successful)
avg_conn = sum(r.connection_time_ms for r in successful) / len(successful)
all_latencies = [l for r in successful for l in r.inter_event_latencies]
all_latencies.sort()
print(f"\nAvg connection time: {avg_conn:.1f}ms")
print(f"Avg time to first event: {avg_ttfe:.1f}ms")
if all_latencies:
p50 = all_latencies[len(all_latencies) // 2]
p95 = all_latencies[int(len(all_latencies) * 0.95)]
p99 = all_latencies[int(len(all_latencies) * 0.99)]
print(f"Inter-event latency p50: {p50:.1f}ms")
print(f"Inter-event latency p95: {p95:.1f}ms")
print(f"Inter-event latency p99: {p99:.1f}ms")
if __name__ == '__main__':
import sys
url = sys.argv[1] iflen(sys.argv) > 1else'http://localhost:3000/events'
results = asyncio.run(load_test_streams(url, concurrent_consumers=50))
print_report(results)
Best Practices
Always set timeouts on stream consumers -- A test that waits forever for a stream event blocks the entire suite.
Use AbortController for fetch-based streams -- Clean cancellation prevents resource leaks in tests.
Validate intermediate state, not just final state -- Streaming is about the journey; assert on each chunk.
Buffer partial data correctly -- Chunks can split across read boundaries; always use a line buffer.
Test empty streams -- A stream that opens and immediately closes should not crash the consumer.
Measure time-to-first-byte separately -- TTFB is the most critical streaming performance metric.
Test with realistic payload sizes -- Small test payloads may miss backpressure and buffering issues.
Close streams in afterEach/afterAll -- Leaked connections cause flaky tests and port exhaustion.
Test the [DONE] signal -- For LLM streams, verify the termination protocol is handled correctly.
Use mock servers, not production endpoints -- Tests must be deterministic; real streaming services are not.
Anti-Patterns to Avoid
Collecting entire stream before asserting -- This defeats the purpose of testing streaming; validate incrementally.
Using setTimeout as synchronization -- Use event-driven assertions (on data, on end) instead of arbitrary delays.
Ignoring partial reads -- A single reader.read() call may not return a complete event; always buffer.
Not testing connection drops -- Real networks fail; simulate disconnections and verify recovery.
Hardcoding port numbers -- Use port 0 and let the OS assign a free port to avoid conflicts.
Skipping error event testing -- The SSE onerror and gRPC on('error') handlers need test coverage.
Testing only happy path timing -- Measure latency under load, not just with a single consumer.
Forgetting to drain the stream -- If a test does not consume the full stream, it may leave the server hanging.
Not validating Content-Type headers -- text/event-stream for SSE is required; wrong headers cause silent failures.
Sharing server instances across parallel tests -- Each test should have its own server to avoid interference.
Running Tests
# Run all streaming tests
npx vitest run tests/sse/ tests/grpc/ tests/chunked/ tests/llm/
# Run SSE tests only
npx vitest run tests/sse/
# Run gRPC streaming tests
npx vitest run tests/grpc/
# Run LLM streaming tests
npx vitest run tests/llm/
# Run with verbose timing output
npx vitest run tests/ --reporter=verbose
# Run performance load test (Python)
python3 tests/performance/stream_load_test.py http://localhost:3000/events
# Run with coverage
npx vitest run tests/ --coverage
# Watch mode for development
npx vitest watch tests/sse/
# Debug a specific test
npx vitest run tests/llm/llm-stream.test.ts --reporter=verbose
