with client.websocket_connect("/ws") as websocket:
            message = "Message one"
            websocket.send_text(message)
            data = websocket.receive_text()
            assert data == f"Message text was: {message}"
            message = "Message two"
            websocket.send_text(message)
            data = websocket.receive_text()

    await websocket.accept()  # pragma: no cover
    await websocket.send_text(f"Item: {item_id}")  # pragma: no cover
    await websocket.close()  # pragma: no cover


@sub_app.websocket("/ws/{item_id}")
async def subapp_websocket_endpoint(websocket: WebSocket, item_id: int):
    await websocket.accept()  # pragma: no cover
    await websocket.send_text(f"Item: {item_id}")  # pragma: no cover
    await websocket.close()  # pragma: no cover

def test_websocket_handle_disconnection(client: TestClient):
    with (
        client.websocket_connect("/ws/1234") as connection,
        client.websocket_connect("/ws/5678") as connection_two,
    ):
        connection.send_text("Hello from 1234")
        data1 = connection.receive_text()
        assert data1 == "You wrote: Hello from 1234"
        data2 = connection_two.receive_text()
        client1_says = "Client #1234 says: Hello from 1234"

            - send
            - accept
            - receive_text
            - receive_bytes
            - receive_json
            - iter_text
            - iter_bytes
            - iter_json
            - send_text
            - send_bytes
            - send_json
            - close

When a client disconnects, a `WebSocketDisconnect` exception is raised, you can catch it.

You can import it directly form `fastapi`:

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    while True:
        data = await websocket.receive_text()

    name: str
    owner: User


@app.post("/users/", response_model=User)
async def create_user(user: UserCreate):
    return user


@app.get("/pets/{pet_id}", response_model=PetOut)
async def read_pet(pet_id: int):
    user = UserDB(
        email="******@****.***",
        hashed_password="secrethashed",
    )
    pet = PetDB(name="Nibbler", owner=user)
    return pet

  }

  @Test
  fun urlConnection() {
    val conn = engine.openConnection(URL("https://google.com/robots.txt")) as HttpURLConnection

    val text =
      conn.inputStream.use {
        it.bufferedReader().readText()
      }

    assertEquals(200, conn.responseCode)

    assertTrue(text.contains("Disallow"))
  }

     * </p>
     *
     * @param reader
     *            the character input stream to read from (must not be {@literal null})
     * @return the text read from the reader
     */
    public static String readText(final Reader reader) {
        assertArgumentNotNull("reader", reader);

        final StringBuilder buf = new StringBuilder(BUF_SIZE);
        CopyUtil.copy(reader, buf);
        return new String(buf);
    }

        await websocket.accept()
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f"Session cookie is: {session}")
            await websocket.send_text(f"Message text was: {data}, for item ID: {item_id}")
    ```
    """

    def __init__(
        self,
        code: Annotated[
            int,
            Doc(

    name: str
    owner: UserBase


@app.post("/users/", response_model=UserBase)
async def create_user(user: UserCreate):
    return user


@app.get("/pets/{pet_id}", response_model=PetOut)
async def read_pet(pet_id: int):
    user = UserDB(
        email="******@****.***",
        hashed_password="secrethashed",
    )
    pet = PetDB(name="Nibbler", owner=user)
    return pet