opentelemetry-test-utils: add async base test class for asgi (#4106)

Author: xrmx

tests/opentelemetry-test-utils/src/opentelemetry/test/asgitestutil.py

@@ -13,6 +13,7 @@
 # limitations under the License.
 
 import asyncio
+from unittest import IsolatedAsyncioTestCase
 
 from asgiref.testing import ApplicationCommunicator
 
@@ -74,3 +75,39 @@ def get_all_output(self):
             except asyncio.TimeoutError:
                 break
         return outputs
+
+
+class AsyncAsgiTestBase(TestBase, IsolatedAsyncioTestCase):
+    def setUp(self):
+        super().setUp()
+
+        self.scope = {}
+        setup_testing_defaults(self.scope)
+        self.communicator = None
+
+    def tearDown(self):
+        if self.communicator:
+            asyncio.get_event_loop().run_until_complete(
+                self.communicator.wait()
+            )
+
+    def seed_app(self, app):
+        self.communicator = ApplicationCommunicator(app, self.scope)
+
+    async def send_input(self, message):
+        await self.communicator.send_input(message)
+
+    async def send_default_request(self):
+        await self.send_input({"type": "http.request", "body": b""})
+
+    async def get_output(self, timeout=0.01):
+        return await self.communicator.receive_output(timeout)
+
+    async def get_all_output(self, timeout=0.01):
+        outputs = []
+        while True:
+            try:
+                outputs.append(await self.communicator.receive_output(timeout))
+            except asyncio.TimeoutError:
+                break
+        return outputs