Test Streaming Code

Testing Something That Arrives in Pieces

A streamed response (Lessons 51 to 60) is not a single value. It is a sequence of events that arrive over time: a start event, a series of text chunks, and a stop event. An application that displays the text as it arrives, or assembles it into a final answer, must handle every event, and it must handle a stream that stops early. The stop may come from a dropped connection, a timeout, or a server-side error, and any of them leaves the caller holding part of an answer.

Testing streaming code is therefore a matter of testing a sequence. The tests must feed the code a chosen sequence of events, including the unusual ones, and check what the code does at each step and at the end. Because a live stream is slow and unpredictable, the tests use a fake stream, built from a generator, that yields the events the test chooses. This lesson covers how to build such a fake, how to test the accumulation and the early-stop cases, and how to check that the stream is closed.

The Code Under Test

The function below consumes a stream of events, collects the text, and reports whether the stream completed. It is the streaming counterpart of the fallback in Lesson 183, with the same principle: the caller always receives a result, even when the stream fails:

"""stream_collect.py: gather text from a stream of events, and report whether it completed."""

from anthropic import APIConnectionError


def collect_text(events) -> tuple[str, bool]:
    parts: list[str] = []
    try:
        for event in events:
            if event.type == "text":
                parts.append(event.text)
            elif event.type == "stop":
                return "".join(parts), True
    except APIConnectionError:
        return "".join(parts), False
    return "".join(parts), False

The function returns a pair: the text received so far, and whether the stop event arrived. A stream that ends without a stop event is incomplete, even though no error was raised, and the function treats it that way.

A Fake Stream

A fake stream is a generator that yields events the test defines. Events are represented by simple objects with a type and the fields the code reads. A generator can also raise an exception partway through, which is how the test produces a dropped connection:

"""stream_demo.py: run the collector against fake streams, including ones that fail."""

from types import SimpleNamespace

import httpx
from anthropic import APIConnectionError

from stream_collect import collect_text

REQUEST = httpx.Request("POST", "https://api.anthropic.com/v1/messages")


def text(chunk):
    return SimpleNamespace(type="text", text=chunk)


STOP = SimpleNamespace(type="stop")


def normal_stream():
    yield text("Hel")
    yield text("lo")
    yield STOP


def dropped_stream():
    yield text("Hel")
    raise APIConnectionError(request=REQUEST)


def unfinished_stream():
    yield text("Hel")
    yield text("lo")


if __name__ == "__main__":
    print("normal:", collect_text(normal_stream()))
    print("dropped:", collect_text(dropped_stream()))
    print("no stop:", collect_text(unfinished_stream()))

Output:

normal: ('Hello', True)
dropped: ('Hel', False)
no stop: ('Hello', False)

How It Works

1. The normal stream yields two chunks and a stop, and the collector joins the chunks into the full word. The completion flag is true because the stop event arrived.
2. The dropped stream raises after one chunk. The collector keeps the chunk it received, which is the partial answer that Lesson 145 said should be preserved, and reports that the stream did not complete.
3. The unfinished stream ends quietly, with no stop event and no error. This case is easy to miss, because nothing failed, yet the answer is still incomplete. The test ensures the flag is false.
4. The three cases cover the three outcomes a stream can have: complete, interrupted, and silently truncated.

Writing the Tests

The three cases become tests that assert on the pair the collector returns. Each one names the behavior it protects, so a failure points directly to the broken rule:

"""test_stream_collect.py: the collector keeps partial text and reports completion accurately."""

from types import SimpleNamespace

import httpx
from anthropic import APIConnectionError

from stream_collect import collect_text

REQUEST = httpx.Request("POST", "https://api.anthropic.com/v1/messages")
STOP = SimpleNamespace(type="stop")


def chunks(*texts):
    return [SimpleNamespace(type="text", text=t) for t in texts]


def test_complete_stream_is_reported_complete():
    assert collect_text(chunks("Hel", "lo") + [STOP]) == ("Hello", True)


def test_dropped_stream_keeps_partial_text():
    def events():
        yield from chunks("Hel")
        raise APIConnectionError(request=REQUEST)
    assert collect_text(events()) == ("Hel", False)


def test_stream_without_stop_is_incomplete():
    assert collect_text(chunks("Hel", "lo")) == ("Hello", False)

Running python -m pytest -q test_stream_collect.py reports three passed tests. The first test passes a list rather than a generator, which the collector accepts because it only iterates over its input.

Checking the Stream Is Closed

A real stream holds a connection open until it is closed. An application that exits early, because of an error or a limit, must close the stream, or the connection stays open until a timeout. The SDK's streaming interface is a context manager, which closes the stream when the block ends, even if an exception occurred. A test can check that behavior with a fake context manager that records whether it was closed:

class FakeStream:
    def __init__(self, events):
        self.events = events
        self.closed = False

    def __enter__(self):
        return self

    def __exit__(self, *exc):
        self.closed = True
        return False

    def __iter__(self):
        return iter(self.events)


stream = FakeStream(chunks("Hel") + [])
with stream as open_stream:
    collect_text(open_stream)
print("closed:", stream.closed)

Output:

closed: True

The check confirms that leaving the block closes the stream, which the test can rely on because the context manager's __exit__ is always called. The fake records the closure in an attribute, so the test asserts on it directly.

Testing the Application Around the Stream

The collector is one piece. The application that displays text as it arrives, or saves the final answer, needs its own tests. The most useful ones check that displayed text matches the collected text, and that a partial answer is labeled as partial to the user. A test that feeds a dropped stream should confirm that the interface shows the partial text with a notice, not a finished-looking answer.

Summary

A stream is a sequence of events that can end in three ways: completed, interrupted by an error, or silently truncated. Tests feed the collector a chosen sequence, using generators that yield events and can raise partway through, and assert on the text and the completion flag. A fake context manager confirms that the stream is closed when the code leaves the block. The application around the collector needs its own tests, so that the user sees partial answers as partial. The next lesson turns to measuring the tokens an application uses, which is the foundation for controlling its cost.