Added tests for ASGI

Signed-off-by: Emil Madsen <sovende@gmail.com>

prometheus_client/asgi.py

@@ -25,7 +25,7 @@ async def prometheus_app(scope, receive, send):
                     "type": "http.response.start",
                     "status": int(status.split(' ')[0]),
                     "headers": [
-                        (x.encode('utf8') for x in header)
+                        tuple(x.encode('utf8') for x in header)
                     ]
                 }
             )

tests/test_asgi.py

@@ -0,0 +1,118 @@
+from __future__ import absolute_import, unicode_literals
+
+import sys
+from unittest import TestCase
+from parameterized import parameterized
+
+from prometheus_client import CollectorRegistry, Counter, generate_latest
+from prometheus_client.exposition import CONTENT_TYPE_LATEST
+
+if sys.version_info < (2, 7):
+    from unittest2 import skipUnless
+else:
+    from unittest import skipUnless
+
+try:
+    # Python >3.5 only
+    from prometheus_client import make_asgi_app
+    import asyncio
+    from asgiref.testing import ApplicationCommunicator
+    HAVE_ASYNCIO_AND_ASGI = True
+except ImportError:
+    HAVE_ASYNCIO_AND_ASGI = False
+
+
+def setup_testing_defaults(scope):
+    scope.update(
+        {
+            "client": ("127.0.0.1", 32767),
+            "headers": [],
+            "http_version": "1.0",
+            "method": "GET",
+            "path": "/",
+            "query_string": b"",
+            "scheme": "http",
+            "server": ("127.0.0.1", 80),
+            "type": "http",
+        }
+    )
+
+
+class ASGITest(TestCase):
+    @skipUnless(HAVE_ASYNCIO_AND_ASGI, "Don't have asyncio/asgi installed.")
+    def setUp(self):
+        self.registry = CollectorRegistry()
+        self.captured_status = None
+        self.captured_headers = None
+        # Setup ASGI scope
+        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)
+
+    def send_input(self, payload):
+        asyncio.get_event_loop().run_until_complete(
+            self.communicator.send_input(payload)
+        )
+
+    def send_default_request(self):
+        self.send_input({"type": "http.request", "body": b""})
+
+    def get_output(self):
+        output = asyncio.get_event_loop().run_until_complete(
+            self.communicator.receive_output(0)
+        )
+        return output
+
+    def get_all_output(self):
+        outputs = []
+        while True:
+            try:
+                outputs.append(self.get_output())
+            except asyncio.TimeoutError:
+                break
+        return outputs
+
+    @parameterized.expand([
+        ["counter", "A counter", 2],
+        ["counter", "Another counter", 3],
+        ["requests", "Number of requests", 5],
+        ["failed_requests", "Number of failed requests", 7],
+    ])
+    def test_reports_metrics(self, metric_name, help_text, increments):
+        """
+        ASGI app serves the metrics from the provided registry.
+        """
+        c = Counter(metric_name, help_text, registry=self.registry)
+        for _ in range(increments):
+            c.inc()
+        # Create and run ASGI app
+        app = make_asgi_app(self.registry)
+        self.seed_app(app)
+        self.send_default_request()
+        # Assert outputs
+        outputs = self.get_all_output()
+        # Assert outputs
+        self.assertEqual(len(outputs), 2)
+        response_start = outputs[0]
+        self.assertEqual(response_start['type'], 'http.response.start')
+        response_body = outputs[1]
+        self.assertEqual(response_body['type'], 'http.response.body')
+        # Status code
+        self.assertEqual(response_start['status'], 200)
+        # Headers
+        self.assertEqual(len(response_start['headers']), 1)
+        self.assertEqual(response_start['headers'][0], (b"Content-Type", CONTENT_TYPE_LATEST.encode('utf8')))
+        # Body
+        output = response_body['body'].decode('utf8')
+        self.assertIn("# HELP " + metric_name + "_total " + help_text + "\n", output)
+        self.assertIn("# TYPE " + metric_name + "_total counter\n", output)
+        self.assertIn(metric_name + "_total " + str(increments) + ".0\n", output)

tests/test_wsgi.py

@@ -1,49 +1,55 @@
 from __future__ import absolute_import, unicode_literals
 
 import sys
+from unittest import TestCase
+from parameterized import parameterized
+from wsgiref.util import setup_testing_defaults
+from prometheus_client import make_wsgi_app
 
 from prometheus_client import CollectorRegistry, Counter, generate_latest
 from prometheus_client.exposition import CONTENT_TYPE_LATEST
 
-if sys.version_info < (2, 7):
-    from unittest2 import skipUnless
-else:
-    from unittest import skipUnless
-
-from prometheus_client import make_wsgi_app
-from unittest import TestCase
-from wsgiref.util import setup_testing_defaults
-from parameterized import parameterized
-
 
 class WSGITest(TestCase):
     def setUp(self):
         self.registry = CollectorRegistry()
         self.captured_status = None
         self.captured_headers = None
+        # Setup WSGI environment
+        self.environ = {}
+        setup_testing_defaults(self.environ)
 
     def capture(self, status, header):
         self.captured_status = status
         self.captured_headers = header
 
+    def assertIn(self, item, iterable):
+        try:
+            super().assertIn(item, iterable)
+        except:  # Python < 2.7
+            self.assertTrue(
+                item in iterable,
+                msg="{item} not found in {iterable}".format(
+                    item=item, iterable=iterable
+                )
+            )
+
     @parameterized.expand([
-        ["counter", "A counter"],
-        ["counter", "Another counter"],
-        ["requests", "Number of requests"],
-        ["failed_requests", "Number of failed requests"],
+        ["counter", "A counter", 2],
+        ["counter", "Another counter", 3],
+        ["requests", "Number of requests", 5],
+        ["failed_requests", "Number of failed requests", 7],
     ])
-    def test_reports_metrics(self, metric_name, help_text):
+    def test_reports_metrics(self, metric_name, help_text, increments):
         """
         WSGI app serves the metrics from the provided registry.
         """
         c = Counter(metric_name, help_text, registry=self.registry)
-        c.inc()
-        # Setup WSGI environment
-        environ = {}
-        setup_testing_defaults(environ)
+        for _ in range(increments):
+            c.inc()
         # Create and run WSGI app
         app = make_wsgi_app(self.registry)
-        outputs = app(environ, self.capture)
+        outputs = app(self.environ, self.capture)
         # Assert outputs
         self.assertEqual(len(outputs), 1)
         output = outputs[0].decode('utf8')
@@ -55,4 +61,4 @@ def test_reports_metrics(self, metric_name, help_text):
         # Body
         self.assertIn("# HELP " + metric_name + "_total " + help_text + "\n", output)
         self.assertIn("# TYPE " + metric_name + "_total counter\n", output)
-        self.assertIn(metric_name + "_total 1.0\n", output)
+        self.assertIn(metric_name + "_total " + str(increments) + ".0\n", output)

tox.ini

@@ -6,6 +6,7 @@ envlist = coverage-clean,py26,py27,py34,py35,py36,py37,py38,pypy,pypy3,{py27,py3
 deps =
     coverage
     pytest
+    parameterized
 
 [testenv:py26]
 ; Last pytest and py version supported on py26 .
@@ -15,6 +16,7 @@ deps =
   pytest==2.9.2
   coverage
   futures
+  parameterized
 
 [testenv:py27]
 deps =
@@ -30,6 +32,7 @@ deps =
 deps =
     {[base]deps}
     {py27,py37,pypy,pypy3}: twisted
+    {py35,py36,py37,py38,pypy3}: asgiref
 commands = coverage run --parallel -m pytest {posargs}
 
 ; Ensure test suite passes if no optional dependencies are present.