From f1e580b1a14f980bf001662c74b0382e644850be Mon Sep 17 00:00:00 2001 From: Helmut Grohne Date: Sat, 17 Jun 2023 15:10:02 +0200 Subject: 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. --- test.py | 31 +++++++++++++++++++++++++++++++ 1 file changed, 31 insertions(+) (limited to 'test.py') 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) -- cgit v1.2.3