test_stream_method_definition_in_sync() — anthropic-sdk-python Function Reference

Architecture documentation for the test_stream_method_definition_in_sync() function in test_beta_messages.py from the anthropic-sdk-python codebase.

Function python AnthropicClient SyncAPI

Entity Profile

AnthropicClient→ SyncAPI→ test_stream_method_definition_in_sync() — anthropic-sdk-python Function Reference

Dependency Diagram

graph TD
  0718485f_5c19_978f_db8a_520bf6dc32c6["test_stream_method_definition_in_sync()"]
  239f7d64_6d26_d418_04e3_d586fe367986["test_beta_messages.py"]
  0718485f_5c19_978f_db8a_520bf6dc32c6 -->|defined in| 239f7d64_6d26_d418_04e3_d586fe367986
  style 0718485f_5c19_978f_db8a_520bf6dc32c6 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

tests/lib/streaming/test_beta_messages.py lines 376–407

def test_stream_method_definition_in_sync(sync: bool) -> None:
    client: Anthropic | AsyncAnthropic = sync_client if sync else async_client

    sig = inspect.signature(client.beta.messages.stream)
    generated_sig = inspect.signature(client.beta.messages.create)

    errors: list[str] = []

    for name, generated_param in generated_sig.parameters.items():
        if name == "stream":
            # intentionally excluded
            continue

        if name == "output_format":
            continue

        custom_param = sig.parameters.get(name)
        if not custom_param:
            errors.append(f"the `{name}` param is missing")
            continue

        if custom_param.annotation != generated_param.annotation:
            errors.append(
                f"types for the `{name}` param are do not match; generated={repr(generated_param.annotation)} custom={repr(custom_param.annotation)}"
            )
            continue

    if errors:
        raise AssertionError(
            f"{len(errors)} errors encountered with the {'sync' if sync else 'async'} client `messages.stream()` method:\n\n"
            + "\n\n".join(errors)
        )