def make_standalone_server(
    interceptors: t.Sequence[aio.ServerInterceptor] | None = None,
    host: str = "0.0.0.0",
) -> t.Generator[tuple[aio.Server, str], None, None]:
    """
    Create a standalone aio.Server for testing.

    Args:
        interceptors: The interceptors to use for the server, default to None.

    Returns:
        The server and the host_url.

    Example for async test cases:

    .. code-block:: python

        async def test_some_async():
            with make_standalone_server() as (server, host_url):
                try:
                    await server.start()
                    channel = grpc.aio.insecure_channel(host_url)
                    ...  # test code here
                finally:
                    await server.stop(None)

    Example for sync test cases:

    .. code-block:: python

        def test_cases():
            import asyncio

            loop = asyncio.new_event_loop()
            with make_standalone_server() as (server, host_url):
                try:
                    loop.run_until_complete(server.start())
                    channel = grpc.insecure_channel(host_url)
                    ...  # test code here
                finally:
                    loop.call_soon_threadsafe(lambda: asyncio.ensure_future(server.stop(None)))
                    loop.close()
                assert loop.is_closed()
    """
    stack = ExitStack()
    port = stack.enter_context(reserve_free_port(enable_so_reuseport=True))
    server = aio.server(
        interceptors=interceptors,
        options=(("grpc.so_reuseport", 1),),
    )
    server.add_insecure_port(f"{host}:{port}")
    print("Using port %d..." % port)
    try:
        yield server, "localhost:%d" % port
    finally:
        stack.close()
Create a standalone aio.Server for testing.

Args:
interceptors: The interceptors to use for the server, default to None.

Returns:
The server and the host_url.

Example for async test cases:

.. code-block:: python

async def test_some_async():
with make_standalone_server() as (server, host_url):
try:
await server.start()
channel = grpc.aio.insecure_channel(host_url)
...  # test code here
finally:
await server.stop(None)

Example for sync test cases:

.. code-block:: python

def test_cases():
import asyncio

loop = asyncio.new_event_loop()
with make_standalone_server() as (server, host_url):
try:
loop.run_until_complete(server.start())
channel = grpc.insecure_channel(host_url)
...  # test code here
finally:
loop.call_soon_threadsafe(lambda: asyncio.ensure_future(server.stop(None)))
loop.close()
assert loop.is_closed()
src/bentoml/testing/grpc/__init__.py