1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
|
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 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):
pass
async def __call__(self, receive, send):
pass
class DoubleApplicationClassNestedFunction:
"""
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
"""
pass
class SingleApplicationClass:
"""
A single-callable class (where you'd pass the class instance in,
e.g. middleware)
"""
def __init__(self):
pass
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}
|
