Repository: django/asgiref
Pull request #296 (2021-09-10 16:06:57)

add more test of StatelessServer, change the way to get event_loop an…
I add some tests for StatelessServer and change two places.

1. Use asyncio.get_event_loop() to replace get_running_loop() to prevent that there is no runing loop.

2. For python version 3.7,  '''run_future''' works as  ```asyncio.run```,  and then trigger ```set_event_loop(None)```.  Eventually, the error mentioned in [https://github.com/django/asgiref/pull/294#issuecomment-909294999](url) happened.  So I change it to ``` asyncio.ensure_future ```.

```
# compatibility.py
if sys.version_info >= (3, 7):
    # these were introduced in 3.7
    get_running_loop = asyncio.get_running_loop
    run_future = asyncio.run
```

```
# cpython/Lib/asyncio/runners.py 
def run(main, *, debug=None):
    ...
    loop = events.new_event_loop()
    try:
        events.set_event_loop(loop)
        if debug is not None:
            loop.set_debug(debug)
        return loop.run_until_complete(main)
    finally:
        try:
            _cancel_all_tasks(loop)
            loop.run_until_complete(loop.shutdown_asyncgens())
            loop.run_until_complete(loop.shutdown_default_executor())
        finally:
            events.set_event_loop(None)
            loop.close()
```



andrewgodwin: Your code won't work on Python 3.6, which `asgiref` still supports (it's a lot of the reason we use very specific asyncio calls and the compatibility layer). You'll need to fix that before I can fully review it.
bright2227: You can review it now. I already fix the problem and check it on the forked repository.   I apologized that I generated a lot of workflows here because I tested workflow_dispatch of Github actions.  I am a newbie programmer, let me know if there is anything I should notice. Thanks.

Patch:
diff --git a/asgiref/server.py b/asgiref/server.py
index fb1c394..43c28c6 100644
--- a/asgiref/server.py
+++ b/asgiref/server.py
@@ -3,7 +3,7 @@ import logging
 import time
 import traceback
 
-from .compatibility import get_running_loop, guarantee_single_callable, run_future
+from .compatibility import guarantee_single_callable
 
 logger = logging.getLogger(__name__)
 
@@ -56,7 +56,7 @@ class StatelessServer:
         """
         Runs the asyncio event loop with our handler loop.
         """
-        event_loop = get_running_loop()
+        event_loop = asyncio.get_event_loop()
         asyncio.ensure_future(self.application_checker())
         try:
             event_loop.run_until_complete(self.handle())
@@ -88,7 +88,7 @@ class StatelessServer:
         input_queue = asyncio.Queue()
         application_instance = guarantee_single_callable(self.application)
         # Run it, and stash the future for later checking
-        future = run_future(
+        future = asyncio.ensure_future(
             application_instance(
                 scope=scope,
                 receive=input_queue.get,


Test patch:
diff --git a/tests/test_server.py b/tests/test_server.py
index c5208b3..296afad 100644
--- a/tests/test_server.py
+++ b/tests/test_server.py
@@ -1,11 +1,149 @@
+import asyncio
+import socket as sock
+from functools import partial
+
+import pytest
+
 from asgiref.server import StatelessServer
 
 
-def test_stateless_server():
-    """StatelessServer can be instantiated with an ASGI 3 application."""
+async def sock_recvfrom(sock, n):
+    while True:
+        try:
+            return sock.recvfrom(n)
+        except BlockingIOError:
+            await asyncio.sleep(0)
+
+
+class Server(StatelessServer):
+    def __init__(self, application, max_applications=1000):
+        super().__init__(
+            application,
+            max_applications=max_applications,
+        )
+        self._sock = sock.socket(sock.AF_INET, sock.SOCK_DGRAM | sock.SOCK_NONBLOCK)
+        self._sock.bind(("127.0.0.1", 0))
+
+    @property
+    def address(self):
+        return self._sock.getsockname()
+
+    async def handle(self):
+        while True:
+            data, addr = await sock_recvfrom(self._sock, 4096)
+            data = data.decode("utf-8")
+
+            if data.startswith("Register"):
+                _, usr_name = data.split(" ")
+                input_quene = self.get_or_create_application_instance(usr_name, addr)
+                input_quene.put_nowait(b"Welcome")
+
+            elif data.startswith("To"):
+                _, usr_name, msg = data.split(" ", 2)
+                input_quene = self.get_or_create_application_instance(usr_name, addr)
+                input_quene.put_nowait(msg.encode("utf-8"))
+
+    async def application_send(self, scope, message):
+        self._sock.sendto(message, scope)
 
+    def close(self):
+        self._sock.close()
+        for details in self.application_instances.values():
+            details["future"].cancel()
+
+
+class Client:
+    def __init__(self, name):
+        self._sock = sock.socket(sock.AF_INET, sock.SOCK_DGRAM | sock.SOCK_NONBLOCK)
+        self.name = name
+
+    async def register(self, server_addr, name=None):
+        name = name or self.name
+        self._sock.sendto(f"Register {name}".encode("utf-8"), server_addr)
+
+    async def send(self, server_addr, to, msg):
+        self._sock.sendto(f"To {to} {msg}".encode("utf-8"), server_addr)
+
+    async def get_msg(self):
+        msg, server_addr = await sock_recvfrom(self._sock, 4096)
+        return msg, server_addr
+
+    def close(self):
+        self._sock.close()
+
+
+@pytest.fixture(scope="function")
+def server():
     async def app(scope, receive, send):
-        pass
+        while True:
+            msg = await receive()
+            await send(msg)
+
+    server = Server(app, 10)
+    yield server
+    server.close()
+
+
+async def check_client_msg(client, expected_address, expected_msg):
+    msg, server_addr = await asyncio.wait_for(client.get_msg(), timeout=1.0)
+    assert msg == expected_msg
+    assert server_addr == expected_address
+
+
+async def server_auto_close(fut, timeout):
+    """Server run based on run_until_complete. It will block forever with handle
+    function because it is a while True loop without break.  Use this method to close
+    server automatically."""
+    loop = asyncio.get_event_loop()
+    task = asyncio.ensure_future(fut, loop=loop)
+    await asyncio.sleep(timeout)
+    task.cancel()
+
+
+def test_stateless_server(server):
+    """StatelessServer can be instantiated with an ASGI 3 application."""
+    """Create a UDP Server can register instance based on name from message of client.
+    Clients can communicate to other client by name through server"""
+
+    loop = asyncio.get_event_loop()
+    server.handle = partial(server_auto_close, fut=server.handle(), timeout=1.0)
+
+    client1 = Client(name="client1")
+    client2 = Client(name="client2")
+
+    async def check_client1_behavior():
+        await client1.register(server.address)
+        await check_client_msg(client1, server.address, b"Welcome")
+        await client1.send(server.address, "client2", "Hello")
+
+    async def check_client2_behavior():
+        await client2.register(server.address)
+        await check_client_msg(client2, server.address, b"Welcome")
+        await check_client_msg(client2, server.address, b"Hello")
+
+    task1 = loop.create_task(check_client1_behavior())
+    task2 = loop.create_task(check_client2_behavior())
+
+    server.run()
+
+    assert task1.done()
+    assert task2.done()
+
+
+def test_server_delete_instance(server):
+    """The max_applications of Server is 10. After 20 times register, application number should be 10."""
+    loop = asyncio.get_event_loop()
+    server.handle = partial(server_auto_close, fut=server.handle(), timeout=1.0)
+
+    client1 = Client(name="client1")
+
+    async def client1_multiple_register():
+        for i in range(20):
+            await client1.register(server.address, name=f"client{i}")
+            print(f"client{i}")
+            await check_client_msg(client1, server.address, b"Welcome")
+
+    task = loop.create_task(client1_multiple_register())
+    server.run()
 
-    server = StatelessServer(app)
-    server.get_or_create_application_instance("scope_id", {})
+    assert task.done()