test_asgi_interface.py

import asyncio
import inspect
from functools import partial

import pytest

scope = object()
receive = object()
send = object()


class App:
    def __init__(self, scope):
        self.scope = scope
    async def __call__(self, receive, send):
        return self.scope, receive, send


class ClassOldDouble:
    def __init__(self):
        pass
    async def asgi(self, receive, send):
        return self.scope, receive, send
    def __call__(self, scope):
        self.scope = scope
        return self.asgi


def old_double(scope):
    async def _inner(scope, receive, send):
        return scope, receive, send

    return partial(_inner, scope)


class NewClass:
    async def __call__(self, scope, receive, send):
        return scope, receive, send


async def new_def(scope, receive, send):
    return scope, receive, send


@pytest.mark.parametrize("app", [App, ClassOldDouble(), old_double, NewClass(), new_def])
def test_asgi_interface(app):
    double_call = None

    is_class = inspect.isclass(app)
    is_class_instance = inspect.isclass(type(app)) and not inspect.isfunction(app)
    instance_with_async_call = is_class_instance and inspect.iscoroutinefunction(app.__call__)
    instance_with_sync_call = is_class_instance and not inspect.iscoroutinefunction(app.__call__)
    is_async_function = inspect.iscoroutinefunction(app)
    is_sync_function = inspect.isfunction(app) and not inspect.iscoroutinefunction(app)

    if is_class:
        double_call = True
    elif instance_with_async_call:
        double_call = False
    elif instance_with_sync_call:
        double_call = True
    elif is_async_function:
        double_call = False
    elif is_sync_function:
        double_call = True

    assert double_call is not None

    if double_call:
        asgi_instance = app(scope)
        result = asyncio.run(asgi_instance(receive, send))
    else:
        result = asyncio.run(app(scope, receive, send))
    assert result[0] is scope
    assert result[1] is receive
    assert result[2] is send