streaming-api-testing

name	Streaming API Testing
description	Streaming API testing skill covering Server-Sent Events testing, chunked transfer encoding, gRPC streaming, real-time data validation, backpressure testing, connection resilience, and AI/LLM streaming response testing.
version	1.0.0
author	thetestingacademy
license	MIT
tags	["streaming","sse","server-sent-events","grpc-streaming","real-time","chunked-transfer","ai-streaming","backpressure"]
testingTypes	["api","integration","performance"]
frameworks	["playwright","jest","vitest"]
languages	["typescript","javascript","python"]
domains	["api","backend","ai"]
agents	["claude-code","cursor","github-copilot","windsurf","codex","aider","continue","cline","zed","bolt","gemini-cli","amp"]

Streaming API Testing Skill

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

1. Test the stream lifecycle -- Verify connection establishment, data flow, and graceful termination.
2. Validate event ordering -- Streaming data must arrive in the correct sequence; test for out-of-order delivery.
3. Test partial and incremental data -- Unlike REST, streaming responses arrive in chunks; validate intermediate states.
4. Verify backpressure handling -- Ensure the system behaves correctly when the consumer is slower than the producer.
5. Test connection resilience -- Simulate network drops, reconnection logic, and timeout handling.
6. Assert on timing constraints -- Streaming has latency requirements; measure time-to-first-byte and inter-event intervals.
7. Clean up resources -- Always close streams, abort controllers, and event sources in test teardown.

Project Structure

project/
  src/
    api/
      sse-endpoint.ts
      grpc-service.ts
      chunked-endpoint.ts
      llm-stream.ts
    tests/
      sse/
        sse-basic.test.ts
        sse-reconnection.test.ts
        sse-backpressure.test.ts
      grpc/
        server-streaming.test.ts
        client-streaming.test.ts
        bidirectional.test.ts
      chunked/
        chunked-transfer.test.ts
        chunked-json.test.ts
      llm/
        llm-stream.test.ts
        token-validation.test.ts
      helpers/
        stream-collector.ts
        mock-sse-server.ts
        mock-grpc-server.ts
        timing-utils.ts
  vitest.config.ts

SSE Endpoint Testing

Basic SSE Connection and Event Validation

// tests/sse/sse-basic.test.ts
import { describe, it, expect, beforeAll, afterAll, afterEach } from 'vitest';
import { createServer, Server } from 'http';
import EventSource from 'eventsource';

 : ;
 : ;

( () => {
  server = ( {
     (req. === ) {
      res.(, {
        : ,
        : ,
        : ,
      });

       count = ;
       interval = ( {
        count++;
        res.();
        res.();
        res.();

         (count >= ) {
          (interval);
          res.();
        }
      }, );

      req.(,  (interval));
    }
  });

    <>( {
    server.(,  {
       addr = server.();
       ( addr ===  && addr) {
        baseUrl = ;
      }
      ();
    });
  });
});

( {
  server.();
});

(,  {
  (,  () => {
     : [] = [];

      <>( {
       es =  ();
       timeout = ( {
        es.();
        ( ());
      }, );

      es. =  {
        events.(.(event.));
      };

      es. =  {
        (timeout);
        es.();
        (); 
      };
    });

    (events).();
    (events.( e.)).([, , , , ]);
  });

  (,  () => {
     response =  ();

    (response..()).();
    (response..()).();
    (response..()).();

    
     response.?.();
  });

  (,  () => {
     : [] = [];

      <>( {
       es =  ();

      es. =  {
        eventIds.(event.);
         (eventIds. >= ) {
          es.();
          ();
        }
      };

      es. =  {
        es.();
        ();
      };
    });

    (eventIds).([, , , , ]);
  });
});