add a wsgitools.scgi.asyncio module

This adds an asyncio implementation of the server side of the SCGI
protocol, because asyncore is being deprecated. Unlike the asyncore
implementation, this does not yet support sendfile.

1 files changed, 31 insertions, 0 deletions

diff --git a/test.py b/test.py
index 5271819..7d6a463 100755
--- a/test.py
+++ b/test.py
@@ -498,6 +498,36 @@ class ScgiForkTest(unittest.TestCase):
         self.assertTrue(data.startswith(b"Status: 200 OK\r\n"))
         self.assertTrue(data.endswith(b"\r\n\r\nnothing"))
 
+from wsgitools.scgi.asyncio import SCGIProtocolFactory
+import asyncio
+
+class ScgiAsyncioTest(unittest.TestCase):
+    def testSimple(self) -> None:
+        asyncio.get_event_loop().run_until_complete(self.asyncTestSimple())
+
+    async def asyncTestSimple(self) -> None:
+        app = applications.StaticContent(
+            "200 OK", [("Content-Type", "text/plain")], b"nothing"
+        )
+        server = await asyncio.get_running_loop().create_server(
+            SCGIProtocolFactory(app),
+            family=socket.AF_INET,
+            host="localhost",
+            port=0,
+        )
+        port = server.sockets[0].getsockname()[1]
+        reader, writer = await asyncio.open_connection("localhost", port)
+        req = {"CONTENT_LENGTH": "0", "REQUEST_METHOD": "GET"}
+        reqb = str2bytes("".join(map("%s\0%s\0".__mod__, req.items())))
+        writer.write(b"%d:%s," % (len(reqb), reqb))
+        await writer.drain()
+        data = await reader.read()
+        writer.close()
+        await writer.wait_closed()
+        self.assertTrue(data.startswith(b"Status: 200 OK\r\n"))
+        self.assertTrue(data.endswith(b"\r\n\r\nnothing"))
+
+
 def alltests(case):
     return unittest.TestLoader().loadTestsFromTestCase(case)
 
@@ -514,6 +544,7 @@ fullsuite.addTest(alltests(RequestLogWSGIFilterTest))
 fullsuite.addTest(alltests(GzipWSGIFilterTest))
 fullsuite.addTest(alltests(ScgiAsynchronousTest))
 fullsuite.addTest(alltests(ScgiForkTest))
+fullsuite.addTest(alltests(ScgiAsyncioTest))
 
 if __name__ == "__main__":
     runner = unittest.TextTestRunner(verbosity=2)