"""
Test that async streaming endpoints can be cancelled without hanging.

Ref: https://github.com/fastapi/fastapi/issues/14680
"""

from collections.abc import AsyncIterable

import anyio
import pytest
from fastapi import FastAPI
from fastapi.responses import StreamingResponse

pytestmark = [
    pytest.mark.anyio,
    pytest.mark.filterwarnings("ignore::pytest.PytestUnraisableExceptionWarning"),
]


app = FastAPI()

from collections.abc import Generator
from contextlib import contextmanager
from typing import Annotated, Any

import pytest
from fastapi import Depends, FastAPI
from fastapi.responses import StreamingResponse
from fastapi.testclient import TestClient


class Session:
    def __init__(self) -> None:
        self.data = ["foo", "bar", "baz"]
        self.open = True

    def __iter__(self) -> Generator[str, None, None]:

        "1-beta123",
        "1-m2",
        "1-m11",
        "1-rc",
        "1-cr2",
        "1-rc123",
        "1-SNAPSHOT",
        "1",
        "1-sp",
        "1-sp2",
        "1-sp123",
        "1-abc",
        "1-def",
        "1-pom-1",
        "1-1-snapshot",
        "1-1",
        "1-2",
        "1-123"
    };

    private static final String[] VERSIONS_NUMBER = {

        final Map<String, Object> params = new HashMap<>();
        params.put(Constants.JOB_LOG_ID, "test-log-id-abc");
        scheduledJob.start(params);

        assertTrue(launchNowCalled.get());
        assertNotNull(capturedOption.get());
        assertEquals("test-log-id-abc", capturedOption.get().getParameterMap().get(Constants.JOB_LOG_ID));
    }

    @Test

	const keep = 10
	ss := &suffixSaver{keep: keep}
	ss.Write([]byte("abc"))
	if got := string(ss.Suffix()); got != "abc" {
		t.Errorf("got = %q; want abc", got)
	}
	ss.Write([]byte("defghijklmno"))
	if got := string(ss.Suffix()); got != "fghijklmno" {
		t.Errorf("got = %q; want fghijklmno", got)
	}
	if got, want := ss.Size(), int64(len("abc")+len("defghijklmno")); got != want {
		t.Errorf("Size = %d; want %d", got, want)

from collections.abc import Generator
from contextlib import contextmanager
from typing import Annotated, Any

import pytest
from fastapi import Depends, FastAPI, WebSocket
from fastapi.testclient import TestClient


class Session:
    def __init__(self) -> None:
        self.data = ["foo", "bar", "baz"]
        self.open = True

    def __iter__(self) -> Generator[str, None, None]:
        for item in self.data:

 * @author Jens Nyman
 */
@NullUnmarked
@GwtIncompatible
@J2ktIncompatible
public class FakeTimeLimiterTest extends TestCase {

  private static final int DELAY_MS = 50;
  private static final String RETURN_VALUE = "abc";

  private TimeLimiter timeLimiter;

  @Override
  protected void setUp() throws Exception {
    super.setUp();
    timeLimiter = new FakeTimeLimiter();
  }

from collections.abc import AsyncGenerator, Generator

import pytest
from fastapi import Depends, FastAPI
from fastapi.testclient import TestClient

app = FastAPI()


class CallableDependency:
    def __call__(self, value: str) -> str:
        return value


class CallableGenDependency:
    def __call__(self, value: str) -> Generator[str, None, None]:
        yield value


class AsyncCallableDependency:

        void unwrapDOSFilter_returnsInnerDosFileFilter_whenWrapped() throws Exception {
            // Arrange: create a ResourceFilterWrapper that wraps a DosFileFilter
            DosFileFilter dos = new DosFileFilter("abc", 7);
            Object wrapper = newPrivateInner("ResourceFilterWrapper", new Class<?>[] { SmbFileFilter.class }, dos);
            // unwrapDOSFilter is a private static method on SmbEnumerationUtil

        .containsExactly(fullpath("base/relative.jar"));
  }

  public void testGetClassName() {
    assertThat(ClassPath.getClassName("abc/d/Abc.class")).isEqualTo("abc.d.Abc");
  }

  public void testResourceInfo_of() {
    assertThat(resourceInfo(ClassPathTest.class)).isInstanceOf(ClassInfo.class);
    assertThat(resourceInfo(ClassPath.class)).isInstanceOf(ClassInfo.class);