From a840723effd9eed605f09318602339bd02d8ffe2 Mon Sep 17 00:00:00 2001 From: davitacols <98448386+davitacols@users.noreply.github.com> Date: Sat, 23 Dec 2023 19:44:31 +0100 Subject: [PATCH 1/8] Add type annotations to asgiref functions --- asgiref/compatibility.py | 93 +++++++++++++++++++++------------------- 1 file changed, 50 insertions(+), 43 deletions(-) diff --git a/asgiref/compatibility.py b/asgiref/compatibility.py index 3a2a63e6..d0762a2f 100644 --- a/asgiref/compatibility.py +++ b/asgiref/compatibility.py @@ -1,48 +1,55 @@ import inspect +from typing import Callable, Coroutine, Optional, Type from .sync import iscoroutinefunction -def is_double_callable(application): - """ - Tests to see if an application is a legacy-style (double-callable) application. - """ - # Look for a hint on the object first - if getattr(application, "_asgi_single_callable", False): - return False - if getattr(application, "_asgi_double_callable", False): - return True - # Uninstanted classes are double-callable - if inspect.isclass(application): - return True - # Instanted classes depend on their __call__ - if hasattr(application, "__call__"): - # We only check to see if its __call__ is a coroutine function - - # if it's not, it still might be a coroutine function itself. - if iscoroutinefunction(application.__call__): - return False - # Non-classes we just check directly - return not iscoroutinefunction(application) - - -def double_to_single_callable(application): - """ - Transforms a double-callable ASGI application into a single-callable one. - """ - - async def new_application(scope, receive, send): - instance = application(scope) - return await instance(receive, send) - - return new_application - - -def guarantee_single_callable(application): - """ - Takes either a single- or double-callable application and always returns it - in single-callable style. Use this to add backwards compatibility for ASGI - 2.0 applications to your server/test harness/etc. - """ - if is_double_callable(application): - application = double_to_single_callable(application) - return application +def is_double_callable(application: Callable[..., Optional[Callable[[], Coroutine[Any, Any, None]]]]) -> bool: + """ + Tests to see if an application is a legacy-style (double-callable) application. + """ + # Look for a hint on the object first + if getattr(application, "_asgi_single_callable", False): + return False + if getattr(application, "_asgi_double_callable", False): + return True + # Uninstantiated classes are double-callable + if inspect.isclass(application): + return True + # Instantiated classes depend on their __call__ + if hasattr(application, "__call__"): + # We only check to see if its __call__ is a coroutine function - + # if it's not, it still might be a coroutine function itself. + if iscoroutinefunction(application.__call__): + return False + # Non-classes we just check directly + return not iscoroutinefunction(application) + + +def double_to_single_callable( + application: Callable[..., Callable[[], Coroutine[Any, Any, None]]] +) -> Callable[[dict, Callable[[], Any], Callable[[Any], None]], Coroutine[Any, Any, None]]: + """ + Transforms a double-callable ASGI application into a single-callable one. + """ + + async def new_application( + scope: dict, receive: Callable[[], Any], send: Callable[[Any], None] + ) -> None: + instance = application(scope) + await instance(receive, send) + + return new_application + + +def guarantee_single_callable( + application: Callable[..., Optional[Callable[[], Coroutine[Any, Any, None]]]] +) -> Callable[[dict, Callable[[], Any], Callable[[Any], None]], Coroutine[Any, Any, None]]: + """ + Takes either a single- or double-callable application and always returns it + in single-callable style. Use this to add backwards compatibility for ASGI + 2.0 applications to your server/test harness/etc. + """ + if is_double_callable(application): + application = double_to_single_callable(application) + return application From 701b0c7c5db4ea955486df73de1390574818f783 Mon Sep 17 00:00:00 2001 From: davitacols <98448386+davitacols@users.noreply.github.com> Date: Sat, 23 Dec 2023 20:23:43 +0100 Subject: [PATCH 2/8] updated documentation --- asgiref/compatibility.py | 103 ++++++++++++++++++++++++--------------- 1 file changed, 63 insertions(+), 40 deletions(-) diff --git a/asgiref/compatibility.py b/asgiref/compatibility.py index d0762a2f..eb5c8872 100644 --- a/asgiref/compatibility.py +++ b/asgiref/compatibility.py @@ -1,55 +1,78 @@ import inspect -from typing import Callable, Coroutine, Optional, Type +from typing import Callable, Coroutine, Optional, Type, Any from .sync import iscoroutinefunction - def is_double_callable(application: Callable[..., Optional[Callable[[], Coroutine[Any, Any, None]]]]) -> bool: - """ - Tests to see if an application is a legacy-style (double-callable) application. - """ - # Look for a hint on the object first - if getattr(application, "_asgi_single_callable", False): - return False - if getattr(application, "_asgi_double_callable", False): - return True - # Uninstantiated classes are double-callable - if inspect.isclass(application): - return True - # Instantiated classes depend on their __call__ - if hasattr(application, "__call__"): - # We only check to see if its __call__ is a coroutine function - - # if it's not, it still might be a coroutine function itself. - if iscoroutinefunction(application.__call__): - return False - # Non-classes we just check directly - return not iscoroutinefunction(application) + """ + Tests whether an application is a legacy-style (double-callable) application. + + Parameters: + - `application`: The callable object to be tested. + + Returns: + - `True` if the application is a double-callable, otherwise `False`. + """ + + # Look for a hint on the object first + if getattr(application, "_asgi_single_callable", False) or getattr(application, "_asgi_double_callable", False): + return True + # Uninstantiated classes are double-callable + if inspect.isclass(application): + return True + # Instantiated classes depend on their __call__ + if hasattr(application, "__call__") and not iscoroutinefunction(application.__call__): + return True + # Non-classes are checked directly + return not iscoroutinefunction(application) def double_to_single_callable( - application: Callable[..., Callable[[], Coroutine[Any, Any, None]]] + application: Callable[..., Callable[[], Coroutine[Any, Any, None]]] ) -> Callable[[dict, Callable[[], Any], Callable[[Any], None]], Coroutine[Any, Any, None]]: - """ - Transforms a double-callable ASGI application into a single-callable one. - """ + """ + Transforms a double-callable ASGI application into a single-callable one. + + Parameters: + - `application`: The double-callable ASGI application. + + Returns: + - A single-callable ASGI application. - async def new_application( - scope: dict, receive: Callable[[], Any], send: Callable[[Any], None] - ) -> None: - instance = application(scope) - await instance(receive, send) + Example: + ```python + single_callable_app = double_to_single_callable(double_callable_app) + ``` + """ - return new_application + async def new_application( + scope: dict, receive: Callable[[], Any], send: Callable[[Any], None] + ) -> None: + instance = application(scope) + await instance(receive, send) + + return new_application def guarantee_single_callable( - application: Callable[..., Optional[Callable[[], Coroutine[Any, Any, None]]]] + application: Callable[..., Optional[Callable[[], Coroutine[Any, Any, None]]]] ) -> Callable[[dict, Callable[[], Any], Callable[[Any], None]], Coroutine[Any, Any, None]]: - """ - Takes either a single- or double-callable application and always returns it - in single-callable style. Use this to add backwards compatibility for ASGI - 2.0 applications to your server/test harness/etc. - """ - if is_double_callable(application): - application = double_to_single_callable(application) - return application + """ + Takes either a single- or double-callable application and always returns it + in single-callable style. + + Parameters: + - `application`: The single- or double-callable ASGI application. + + Returns: + - A single-callable ASGI application. + + Example: + ```python + guaranteed_single_callable = guarantee_single_callable(some_callable_app) + ``` + """ + + if is_double_callable(application): + application = double_to_single_callable(application) + return application From 3e7989e3f2fef04c730074c1b319ac82ae7ecfb0 Mon Sep 17 00:00:00 2001 From: davitacols <98448386+davitacols@users.noreply.github.com> Date: Sun, 24 Dec 2023 13:29:10 +0100 Subject: [PATCH 3/8] updated test_compatibility --- tests/test_compatibility.py | 62 ++++++++++++------------------------- 1 file changed, 19 insertions(+), 43 deletions(-) diff --git a/tests/test_compatibility.py b/tests/test_compatibility.py index 4164683f..953741fd 100644 --- a/tests/test_compatibility.py +++ b/tests/test_compatibility.py @@ -1,26 +1,18 @@ import pytest - from asgiref.compatibility import double_to_single_callable, is_double_callable from asgiref.testing import ApplicationCommunicator def double_application_function(scope): - """ - A nested function based double-callable application. - """ - + """A nested function-based double-callable application.""" async def inner(receive, send): message = await receive() await send({"scope": scope["value"], "message": message["value"]}) - return inner class DoubleApplicationClass: - """ - A classic class-based double-callable application. - """ - + """A classic class-based double-callable application.""" def __init__(self, scope): pass @@ -29,33 +21,23 @@ async def __call__(self, receive, send): class DoubleApplicationClassNestedFunction: - """ - A function closure inside a class! - """ - + """A function closure inside a class.""" def __init__(self): pass def __call__(self, scope): async def inner(receive, send): pass - return inner async def single_application_function(scope, receive, send): - """ - A single-function single-callable application - """ + """A single-function single-callable application.""" pass class SingleApplicationClass: - """ - A single-callable class (where you'd pass the class instance in, - e.g. middleware) - """ - + """A single-callable class.""" def __init__(self): pass @@ -63,32 +45,26 @@ async def __call__(self, scope, receive, send): pass -def test_is_double_callable(): - """ - Tests that the signature matcher works as expected. - """ - assert is_double_callable(double_application_function) is True - assert is_double_callable(DoubleApplicationClass) is True - assert is_double_callable(DoubleApplicationClassNestedFunction()) is True - assert is_double_callable(single_application_function) is False - assert is_double_callable(SingleApplicationClass()) is False +@pytest.mark.asyncio +async def test_is_double_callable(): + """Test the behavior of is_double_callable function.""" + assert is_double_callable(double_application_function) + assert is_double_callable(DoubleApplicationClass) + assert is_double_callable(DoubleApplicationClassNestedFunction()) + assert not is_double_callable(single_application_function) + assert not is_double_callable(SingleApplicationClass()) -def test_double_to_single_signature(): - """ - Test that the new object passes a signature test. - """ - assert ( - is_double_callable(double_to_single_callable(double_application_function)) - is False - ) +@pytest.mark.asyncio +async def test_double_to_single_callable(): + """Test the behavior of double_to_single_callable function.""" + new_app = double_to_single_callable(double_application_function) + assert not is_double_callable(new_app) @pytest.mark.asyncio async def test_double_to_single_communicator(): - """ - Test that the new application works - """ + """Test the behavior of the new application using ApplicationCommunicator.""" new_app = double_to_single_callable(double_application_function) instance = ApplicationCommunicator(new_app, {"value": "woohoo"}) await instance.send_input({"value": 42}) From 3c7bd107445efa7ceac9d9e254eda4590314fad4 Mon Sep 17 00:00:00 2001 From: davitacols <98448386+davitacols@users.noreply.github.com> Date: Sun, 24 Dec 2023 13:33:35 +0100 Subject: [PATCH 4/8] removed unused arguments --- tests/test_compatibility.py | 40 +++++++++++++++++++++++++++---------- 1 file changed, 30 insertions(+), 10 deletions(-) diff --git a/tests/test_compatibility.py b/tests/test_compatibility.py index 953741fd..d8c548d2 100644 --- a/tests/test_compatibility.py +++ b/tests/test_compatibility.py @@ -1,10 +1,16 @@ +""" +Tests for asgiref compatibility functions. +""" + import pytest from asgiref.compatibility import double_to_single_callable, is_double_callable from asgiref.testing import ApplicationCommunicator -def double_application_function(scope): - """A nested function-based double-callable application.""" +async def double_application_function(scope): + """ + A nested function-based double-callable application. + """ async def inner(receive, send): message = await receive() await send({"scope": scope["value"], "message": message["value"]}) @@ -12,7 +18,9 @@ async def inner(receive, send): class DoubleApplicationClass: - """A classic class-based double-callable application.""" + """ + A classic class-based double-callable application. + """ def __init__(self, scope): pass @@ -21,7 +29,9 @@ async def __call__(self, receive, send): class DoubleApplicationClassNestedFunction: - """A function closure inside a class.""" + """ + A function closure inside a class. + """ def __init__(self): pass @@ -32,22 +42,28 @@ async def inner(receive, send): async def single_application_function(scope, receive, send): - """A single-function single-callable application.""" + """ + A single-function single-callable application. + """ pass class SingleApplicationClass: - """A single-callable class.""" + """ + A single-callable class. + """ def __init__(self): pass - async def __call__(self, scope, receive, send): + async def __call__(self, receive, send): pass @pytest.mark.asyncio async def test_is_double_callable(): - """Test the behavior of is_double_callable function.""" + """ + Test the behavior of is_double_callable function. + """ assert is_double_callable(double_application_function) assert is_double_callable(DoubleApplicationClass) assert is_double_callable(DoubleApplicationClassNestedFunction()) @@ -57,14 +73,18 @@ async def test_is_double_callable(): @pytest.mark.asyncio async def test_double_to_single_callable(): - """Test the behavior of double_to_single_callable function.""" + """ + Test the behavior of double_to_single_callable function. + """ new_app = double_to_single_callable(double_application_function) assert not is_double_callable(new_app) @pytest.mark.asyncio async def test_double_to_single_communicator(): - """Test the behavior of the new application using ApplicationCommunicator.""" + """ + Test the behavior of the new application using ApplicationCommunicator. + """ new_app = double_to_single_callable(double_application_function) instance = ApplicationCommunicator(new_app, {"value": "woohoo"}) await instance.send_input({"value": 42}) From 72cd5b8a2ef449510a2b98c03d593928721a9156 Mon Sep 17 00:00:00 2001 From: davitacols <98448386+davitacols@users.noreply.github.com> Date: Sun, 24 Dec 2023 13:54:20 +0100 Subject: [PATCH 5/8] running pylint locally --- tests/test_compatibility.py | 29 ++++++++++++++++++++++------- 1 file changed, 22 insertions(+), 7 deletions(-) diff --git a/tests/test_compatibility.py b/tests/test_compatibility.py index d8c548d2..c05be506 100644 --- a/tests/test_compatibility.py +++ b/tests/test_compatibility.py @@ -21,10 +21,10 @@ class DoubleApplicationClass: """ A classic class-based double-callable application. """ - def __init__(self, scope): + def __init__(self, _): pass - async def __call__(self, receive, send): + async def __call__(self, _, __, ___): pass @@ -35,17 +35,16 @@ class DoubleApplicationClassNestedFunction: def __init__(self): pass - def __call__(self, scope): - async def inner(receive, send): + def __call__(self, _): + async def inner(_, __): pass return inner -async def single_application_function(scope, receive, send): +async def single_application_function(_, __, ___): """ A single-function single-callable application. """ - pass class SingleApplicationClass: @@ -55,7 +54,7 @@ class SingleApplicationClass: def __init__(self): pass - async def __call__(self, receive, send): + async def __call__(self, _, __): pass @@ -89,3 +88,19 @@ async def test_double_to_single_communicator(): instance = ApplicationCommunicator(new_app, {"value": "woohoo"}) await instance.send_input({"value": 42}) assert await instance.receive_output() == {"scope": "woohoo", "message": 42} + + +@pytest.mark.asyncio +async def test_another_feature(): + """ + Test another feature or behavior. + """ + # Add your test logic here + + +@pytest.mark.asyncio +async def test_yet_another_feature(): + """ + Test yet another feature or behavior. + """ + # Add your test logic here From b202201e59785579887373e0727ff0a9c9d73df8 Mon Sep 17 00:00:00 2001 From: davitacols <98448386+davitacols@users.noreply.github.com> Date: Sun, 24 Dec 2023 17:36:28 +0100 Subject: [PATCH 6/8] passed --- tests/test_compatibility.py | 67 +++++++++---------------------------- 1 file changed, 16 insertions(+), 51 deletions(-) diff --git a/tests/test_compatibility.py b/tests/test_compatibility.py index c05be506..953741fd 100644 --- a/tests/test_compatibility.py +++ b/tests/test_compatibility.py @@ -1,16 +1,10 @@ -""" -Tests for asgiref compatibility functions. -""" - import pytest from asgiref.compatibility import double_to_single_callable, is_double_callable from asgiref.testing import ApplicationCommunicator -async def double_application_function(scope): - """ - A nested function-based double-callable application. - """ +def double_application_function(scope): + """A nested function-based double-callable application.""" async def inner(receive, send): message = await receive() await send({"scope": scope["value"], "message": message["value"]}) @@ -18,51 +12,42 @@ async def inner(receive, send): class DoubleApplicationClass: - """ - A classic class-based double-callable application. - """ - def __init__(self, _): + """A classic class-based double-callable application.""" + def __init__(self, scope): pass - async def __call__(self, _, __, ___): + async def __call__(self, receive, send): pass class DoubleApplicationClassNestedFunction: - """ - A function closure inside a class. - """ + """A function closure inside a class.""" def __init__(self): pass - def __call__(self, _): - async def inner(_, __): + def __call__(self, scope): + async def inner(receive, send): pass return inner -async def single_application_function(_, __, ___): - """ - A single-function single-callable application. - """ +async def single_application_function(scope, receive, send): + """A single-function single-callable application.""" + pass class SingleApplicationClass: - """ - A single-callable class. - """ + """A single-callable class.""" def __init__(self): pass - async def __call__(self, _, __): + async def __call__(self, scope, receive, send): pass @pytest.mark.asyncio async def test_is_double_callable(): - """ - Test the behavior of is_double_callable function. - """ + """Test the behavior of is_double_callable function.""" assert is_double_callable(double_application_function) assert is_double_callable(DoubleApplicationClass) assert is_double_callable(DoubleApplicationClassNestedFunction()) @@ -72,35 +57,15 @@ async def test_is_double_callable(): @pytest.mark.asyncio async def test_double_to_single_callable(): - """ - Test the behavior of double_to_single_callable function. - """ + """Test the behavior of double_to_single_callable function.""" new_app = double_to_single_callable(double_application_function) assert not is_double_callable(new_app) @pytest.mark.asyncio async def test_double_to_single_communicator(): - """ - Test the behavior of the new application using ApplicationCommunicator. - """ + """Test the behavior of the new application using ApplicationCommunicator.""" new_app = double_to_single_callable(double_application_function) instance = ApplicationCommunicator(new_app, {"value": "woohoo"}) await instance.send_input({"value": 42}) assert await instance.receive_output() == {"scope": "woohoo", "message": 42} - - -@pytest.mark.asyncio -async def test_another_feature(): - """ - Test another feature or behavior. - """ - # Add your test logic here - - -@pytest.mark.asyncio -async def test_yet_another_feature(): - """ - Test yet another feature or behavior. - """ - # Add your test logic here From 521141e878be2aaf475576c899a3e4a52910e007 Mon Sep 17 00:00:00 2001 From: davitacols <98448386+davitacols@users.noreply.github.com> Date: Sun, 24 Dec 2023 20:12:43 +0100 Subject: [PATCH 7/8] linting --- tests/test_compatibility.py | 76 +++++++++++++++++++++---------------- 1 file changed, 44 insertions(+), 32 deletions(-) diff --git a/tests/test_compatibility.py b/tests/test_compatibility.py index 953741fd..7f33e5f7 100644 --- a/tests/test_compatibility.py +++ b/tests/test_compatibility.py @@ -1,71 +1,83 @@ +""" +Module for testing ASGI compatibility functions and classes. +""" + import pytest from asgiref.compatibility import double_to_single_callable, is_double_callable from asgiref.testing import ApplicationCommunicator +@pytest.mark.asyncio +async def test_is_double_callable(): + """Test the behavior of is_double_callable function.""" + assert is_double_callable(double_application_function) + assert is_double_callable(DoubleApplicationClass) + assert is_double_callable(DoubleApplicationClassNestedFunction()) + assert not is_double_callable(single_application_function) + assert not is_double_callable(SingleApplicationClass()) + + +@pytest.mark.asyncio +async def test_double_to_single_callable(): + """Test the behavior of double_to_single_callable function.""" + new_app = double_to_single_callable(double_application_function) + assert not is_double_callable(new_app) + + +@pytest.mark.asyncio +async def test_double_to_single_communicator(): + """Test the behavior of the new application using ApplicationCommunicator.""" + new_app = double_to_single_callable(double_application_function) + instance = ApplicationCommunicator(new_app, {"value": "woohoo"}) + await instance.send_input({"value": 42}) + output = await instance.receive_output() + assert output == {"scope": "woohoo", "message": 42} + + +# Additional functions and classes for testing def double_application_function(scope): """A nested function-based double-callable application.""" + async def inner(receive, send): message = await receive() await send({"scope": scope["value"], "message": message["value"]}) + return inner class DoubleApplicationClass: """A classic class-based double-callable application.""" - def __init__(self, scope): + + def __init__(self): pass - async def __call__(self, receive, send): + async def __call__(self): pass class DoubleApplicationClassNestedFunction: """A function closure inside a class.""" + def __init__(self): pass - def __call__(self, scope): - async def inner(receive, send): + def __call__(self): + async def inner(): pass + return inner -async def single_application_function(scope, receive, send): +async def single_application_function(): """A single-function single-callable application.""" - pass class SingleApplicationClass: """A single-callable class.""" + def __init__(self): pass - async def __call__(self, scope, receive, send): + async def __call__(self): pass - -@pytest.mark.asyncio -async def test_is_double_callable(): - """Test the behavior of is_double_callable function.""" - assert is_double_callable(double_application_function) - assert is_double_callable(DoubleApplicationClass) - assert is_double_callable(DoubleApplicationClassNestedFunction()) - assert not is_double_callable(single_application_function) - assert not is_double_callable(SingleApplicationClass()) - - -@pytest.mark.asyncio -async def test_double_to_single_callable(): - """Test the behavior of double_to_single_callable function.""" - new_app = double_to_single_callable(double_application_function) - assert not is_double_callable(new_app) - - -@pytest.mark.asyncio -async def test_double_to_single_communicator(): - """Test the behavior of the new application using ApplicationCommunicator.""" - new_app = double_to_single_callable(double_application_function) - instance = ApplicationCommunicator(new_app, {"value": "woohoo"}) - await instance.send_input({"value": 42}) - assert await instance.receive_output() == {"scope": "woohoo", "message": 42} From 9fb1fbcb53abc3cb2c0ac366a0a5f7ef3448229c Mon Sep 17 00:00:00 2001 From: davitacols <98448386+davitacols@users.noreply.github.com> Date: Wed, 27 Dec 2023 14:31:34 +0100 Subject: [PATCH 8/8] work in progress --- asgiref/compatibility.py | 67 ++++++++++--------------- tests/test_compatibility.py | 98 +++++++++++++++++++++---------------- 2 files changed, 81 insertions(+), 84 deletions(-) diff --git a/asgiref/compatibility.py b/asgiref/compatibility.py index eb5c8872..2a4bc29f 100644 --- a/asgiref/compatibility.py +++ b/asgiref/compatibility.py @@ -1,52 +1,45 @@ import inspect -from typing import Callable, Coroutine, Optional, Type, Any +from typing import Any, Callable, Coroutine, Optional from .sync import iscoroutinefunction -def is_double_callable(application: Callable[..., Optional[Callable[[], Coroutine[Any, Any, None]]]]) -> bool: - """ - Tests whether an application is a legacy-style (double-callable) application. - - Parameters: - - `application`: The callable object to be tested. - Returns: - - `True` if the application is a double-callable, otherwise `False`. +def is_double_callable( + application: Callable[..., Optional[Callable[[], Coroutine[Any, Any, None]]]] +) -> bool: + """ + Tests to see if an application is a legacy-style (double-callable) application. """ - # Look for a hint on the object first - if getattr(application, "_asgi_single_callable", False) or getattr(application, "_asgi_double_callable", False): + if getattr(application, "_asgi_single_callable", False): + return False + if getattr(application, "_asgi_double_callable", False): return True # Uninstantiated classes are double-callable if inspect.isclass(application): return True # Instantiated classes depend on their __call__ - if hasattr(application, "__call__") and not iscoroutinefunction(application.__call__): - return True - # Non-classes are checked directly + if hasattr(application, "__call__"): + # We only check to see if its __call__ is a coroutine function - + # if it's not, it still might be a coroutine function itself. + if iscoroutinefunction(application.__call__): + return False + # Non-classes we just check directly return not iscoroutinefunction(application) def double_to_single_callable( application: Callable[..., Callable[[], Coroutine[Any, Any, None]]] -) -> Callable[[dict, Callable[[], Any], Callable[[Any], None]], Coroutine[Any, Any, None]]: +) -> Callable[ + [dict[str, Any], Callable[[], Any], Callable[[Any], None]], + Coroutine[Any, Any, None], +]: """ Transforms a double-callable ASGI application into a single-callable one. - - Parameters: - - `application`: The double-callable ASGI application. - - Returns: - - A single-callable ASGI application. - - Example: - ```python - single_callable_app = double_to_single_callable(double_callable_app) - ``` """ async def new_application( - scope: dict, receive: Callable[[], Any], send: Callable[[Any], None] + scope: dict[str, Any], receive: Callable[[], Any], send: Callable[[Any], None] ) -> None: instance = application(scope) await instance(receive, send) @@ -56,23 +49,15 @@ async def new_application( def guarantee_single_callable( application: Callable[..., Optional[Callable[[], Coroutine[Any, Any, None]]]] -) -> Callable[[dict, Callable[[], Any], Callable[[Any], None]], Coroutine[Any, Any, None]]: +) -> Callable[ + [dict[str, Any], Callable[[], Any], Callable[[Any], None]], + Coroutine[Any, Any, None], +]: """ Takes either a single- or double-callable application and always returns it - in single-callable style. - - Parameters: - - `application`: The single- or double-callable ASGI application. - - Returns: - - A single-callable ASGI application. - - Example: - ```python - guaranteed_single_callable = guarantee_single_callable(some_callable_app) - ``` + in single-callable style. Use this to add backward compatibility for ASGI + 2.0 applications to your server/test harness/etc. """ - if is_double_callable(application): application = double_to_single_callable(application) return application diff --git a/tests/test_compatibility.py b/tests/test_compatibility.py index 7f33e5f7..4164683f 100644 --- a/tests/test_compatibility.py +++ b/tests/test_compatibility.py @@ -1,42 +1,13 @@ -""" -Module for testing ASGI compatibility functions and classes. -""" - import pytest + from asgiref.compatibility import double_to_single_callable, is_double_callable from asgiref.testing import ApplicationCommunicator -@pytest.mark.asyncio -async def test_is_double_callable(): - """Test the behavior of is_double_callable function.""" - assert is_double_callable(double_application_function) - assert is_double_callable(DoubleApplicationClass) - assert is_double_callable(DoubleApplicationClassNestedFunction()) - assert not is_double_callable(single_application_function) - assert not is_double_callable(SingleApplicationClass()) - - -@pytest.mark.asyncio -async def test_double_to_single_callable(): - """Test the behavior of double_to_single_callable function.""" - new_app = double_to_single_callable(double_application_function) - assert not is_double_callable(new_app) - - -@pytest.mark.asyncio -async def test_double_to_single_communicator(): - """Test the behavior of the new application using ApplicationCommunicator.""" - new_app = double_to_single_callable(double_application_function) - instance = ApplicationCommunicator(new_app, {"value": "woohoo"}) - await instance.send_input({"value": 42}) - output = await instance.receive_output() - assert output == {"scope": "woohoo", "message": 42} - - -# Additional functions and classes for testing def double_application_function(scope): - """A nested function-based double-callable application.""" + """ + A nested function based double-callable application. + """ async def inner(receive, send): message = await receive() @@ -46,38 +17,79 @@ async def inner(receive, send): class DoubleApplicationClass: - """A classic class-based double-callable application.""" + """ + A classic class-based double-callable application. + """ - def __init__(self): + def __init__(self, scope): pass - async def __call__(self): + async def __call__(self, receive, send): pass class DoubleApplicationClassNestedFunction: - """A function closure inside a class.""" + """ + A function closure inside a class! + """ def __init__(self): pass - def __call__(self): - async def inner(): + def __call__(self, scope): + async def inner(receive, send): pass return inner -async def single_application_function(): - """A single-function single-callable application.""" +async def single_application_function(scope, receive, send): + """ + A single-function single-callable application + """ + pass class SingleApplicationClass: - """A single-callable class.""" + """ + A single-callable class (where you'd pass the class instance in, + e.g. middleware) + """ def __init__(self): pass - async def __call__(self): + async def __call__(self, scope, receive, send): pass + +def test_is_double_callable(): + """ + Tests that the signature matcher works as expected. + """ + assert is_double_callable(double_application_function) is True + assert is_double_callable(DoubleApplicationClass) is True + assert is_double_callable(DoubleApplicationClassNestedFunction()) is True + assert is_double_callable(single_application_function) is False + assert is_double_callable(SingleApplicationClass()) is False + + +def test_double_to_single_signature(): + """ + Test that the new object passes a signature test. + """ + assert ( + is_double_callable(double_to_single_callable(double_application_function)) + is False + ) + + +@pytest.mark.asyncio +async def test_double_to_single_communicator(): + """ + Test that the new application works + """ + new_app = double_to_single_callable(double_application_function) + instance = ApplicationCommunicator(new_app, {"value": "woohoo"}) + await instance.send_input({"value": 42}) + assert await instance.receive_output() == {"scope": "woohoo", "message": 42}