ServerSentEvent(data="test", retry=-1)


def test_server_sent_event_float_retry_rejected():
    with pytest.raises(ValueError):
        ServerSentEvent(data="test", retry=1.5)  # type: ignore[arg-type]


def test_raw_data_sent_without_json_encoding(client: TestClient):
    """raw_data is sent as-is, not JSON-encoded."""
    response = client.get("/items/stream-raw")
    assert response.status_code == 200

        assertEquals(mockHandle, lsarClose.handle, "The handle should be initialized correctly.");

        // Verify that the ptype is set to 0
        assertEquals(0, lsarClose.getPtype(), "The ptype should be set to 0.");

        // Verify that the flags are set to the combination of DCERPC_FIRST_FRAG and DCERPC_LAST_FRAG

    // Base64 encoded Type 1 message: NTLMSSP, Type 1, flags=0x00088207
    // Domain="DOMAIN", Workstation="WORKSTATION"
    private final String type1MessageBase64 = "TlRMTVNTUAABAAAABzIAAAYABgArAAAACwALACAAAABET01BSU5XT1JLU1RBVElPTg==";

    // Base64 encoded Type 3 message: NTLMSSP, Type 3
    // User: "user", Domain: "DOMAIN", Workstation: "WORKSTATION"
    // This is a valid Type 3 message with proper structure

import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;

/**
 * Captures the actual type of {@code T}.
 *
 * @author Ben Yu
 */
abstract class TypeCapture<T> {

  /** Returns the captured type. */
  final Type capture() {
    Type superclass = getClass().getGenericSuperclass();

        for union_arg in union_args:
            if union_arg is type(None):
                continue
            origin_type = get_origin(union_arg) or union_arg
            break
    assert issubclass(origin_type, shared.sequence_types)  # type: ignore[arg-type]
    return shared.sequence_annotation_to_type[origin_type](value)  # type: ignore[no-any-return,index]

      Comparator<? super K> comparator,
      boolean sameComparator,
      Iterable<? extends Entry<? extends K, ? extends V>> entries) {
    // "adding" type params to an array of a raw type should be safe as
    // long as no one can ever cast that same array instance back to a
    // raw type.
    @SuppressWarnings("unchecked")
    Entry<K, V>[] entryArray = (Entry[]) Iterables.toArray(entries, EMPTY_ENTRY_ARRAY);

        assertEquals(0, share1.type, "Share 1 raw type should be 0.");
        assertEquals("Remark 1", share1.remark.trim(), "Share 1 remark should be correct.");

        SmbShareInfo share2 = (SmbShareInfo) response.results[1];
        assertEquals("IPC$", share2.getName().trim(), "Share 2 name should be correct.");
        assertEquals(3, share2.type, "Share 2 raw type should be 3.");

import java.util.Set;
import org.jspecify.annotations.Nullable;

/**
 * An implementation of {@link GraphConnections} for undirected graphs.
 *
 * @author James Sexton
 * @param <N> Node parameter type
 * @param <V> Value parameter type
 */
final class UndirectedGraphConnections<N, V> implements GraphConnections<N, V> {
  private final Map<N, V> adjacentNodeValues;

  private UndirectedGraphConnections(Map<N, V> adjacentNodeValues) {

If each JSON item you want to send back is of type `Item` (a Pydantic model) and it's an async function, you can declare the return type as `AsyncIterable[Item]`:

{* ../../docs_src/stream_json_lines/tutorial001_py310.py ln[1:24] hl[9:11,22] *}

_SSE_EVENT_SCHEMA: dict[str, Any] = {
    "type": "object",
    "properties": {
        "data": {"type": "string"},
        "event": {"type": "string"},
        "id": {"type": "string"},
        "retry": {"type": "integer", "minimum": 0},
    },
}


class EventSourceResponse(StreamingResponse):
    """Streaming response with `text/event-stream` media type.