Honor HTTP methods in make_asgi_app like make_wsgi_app

prometheus_client/asgi.py

@@ -20,8 +20,27 @@ async def prometheus_app(scope, receive, send):
             value.decode("utf8") for (name, value) in scope.get('headers')
             if name.decode("utf8").lower() == 'accept-encoding'
         ])
-        # Bake output
-        status, headers, output = _bake_output(registry, accept_header, accept_encoding_header, params, disable_compression)
+        method = scope.get('method', 'GET')
+        path = scope.get('path', '')
+
+        # Match make_wsgi_app: only GET serves metrics.
+        if method == 'OPTIONS':
+            status = '200 OK'
+            headers = [('Allow', 'OPTIONS,GET')]
+            output = b''
+        elif method != 'GET':
+            status = '405 Method Not Allowed'
+            headers = [('Allow', 'OPTIONS,GET')]
+            output = '# HTTP {}: {}; use OPTIONS or GET\n'.format(status, method).encode()
+        elif path == '/favicon.ico':
+            # Serve empty response for browsers
+            status = '200 OK'
+            headers = []
+            output = b''
+        else:
+            # Note: For backwards compatibility, the URI path for GET is not
+            # constrained to the documented /metrics, but any path is allowed.
+            status, headers, output = _bake_output(registry, accept_header, accept_encoding_header, params, disable_compression)
         formatted_headers = []
         for header in headers:
             formatted_headers.append(tuple(x.encode('utf8') for x in header))

tests/test_asgi.py

@@ -196,6 +196,63 @@ def test_plaintext_encoding(self):
         content_type = self.get_response_header_value('Content-Type').split(";")[0]
         assert content_type == "text/plain"
 
+    def test_options_method(self):
+        """OPTIONS is answered with Allow and does not scrape the registry."""
+        from unittest.mock import patch
+
+        from prometheus_client.exposition import _bake_output
+
+        app = make_asgi_app(self.registry)
+        self.scope["method"] = "OPTIONS"
+        with patch("prometheus_client.asgi._bake_output", side_effect=_bake_output) as mock:
+            self.seed_app(app)
+            self.send_default_request()
+            outputs = self.get_all_output()
+            self.assertEqual(mock.call_count, 0)
+        response_start = outputs[0]
+        response_body = outputs[1]
+        self.assertEqual(response_start["status"], 200)
+        self.assertIn((b"Allow", b"OPTIONS,GET"), response_start["headers"])
+        self.assertEqual(response_body["body"], b"")
+
+    def test_post_method_not_allowed(self):
+        """Non-GET methods are rejected, matching make_wsgi_app."""
+        from unittest.mock import patch
+
+        from prometheus_client.exposition import _bake_output
+
+        app = make_asgi_app(self.registry)
+        self.scope["method"] = "POST"
+        with patch("prometheus_client.asgi._bake_output", side_effect=_bake_output) as mock:
+            self.seed_app(app)
+            self.send_default_request()
+            outputs = self.get_all_output()
+            self.assertEqual(mock.call_count, 0)
+        response_start = outputs[0]
+        response_body = outputs[1]
+        self.assertEqual(response_start["status"], 405)
+        self.assertIn((b"Allow", b"OPTIONS,GET"), response_start["headers"])
+        self.assertIn(b"405 Method Not Allowed", response_body["body"])
+        self.assertIn(b"POST", response_body["body"])
+
+    def test_favicon_path(self):
+        """Browsers requesting /favicon.ico get an empty 200 without scraping."""
+        from unittest.mock import patch
+
+        from prometheus_client.exposition import _bake_output
+
+        app = make_asgi_app(self.registry)
+        self.scope["path"] = "/favicon.ico"
+        with patch("prometheus_client.asgi._bake_output", side_effect=_bake_output) as mock:
+            self.seed_app(app)
+            self.send_default_request()
+            outputs = self.get_all_output()
+            self.assertEqual(mock.call_count, 0)
+        response_start = outputs[0]
+        response_body = outputs[1]
+        self.assertEqual(response_start["status"], 200)
+        self.assertEqual(response_body["body"], b"")
+
     def test_qs_parsing(self):
         """Only metrics that match the 'name[]' query string param appear"""