Follow every redirect status and method in passthrough_redirect_handler

prometheus_client/exposition.py

@@ -99,8 +99,7 @@ def redirect_request(self, req, fp, code, msg, headers, newurl):
         # indicate the method, by monkeypatching this, instead of setting the
         # Request object's method attribute.
         m = getattr(req, "method", req.get_method())
-        if not (code in (301, 302, 303, 307) and m in ("GET", "HEAD")
-                or code in (301, 302, 303) and m in ("POST", "PUT")):
+        if code not in (301, 302, 303, 307, 308):
             raise HTTPError(req.full_url, code, msg, headers, fp)
         new_request = Request(
             newurl.replace(' ', '%20'),  # space escaping in new url if needed.
@@ -112,6 +111,9 @@ def redirect_request(self, req, fp, code, msg, headers, newurl):
         new_request.method = m
         return new_request
 
+    # urllib only routes a 308 to http_error_308 from Python 3.11 on.
+    http_error_308 = HTTPRedirectHandler.http_error_302
+
 
 def _bake_output(registry, accept_header, accept_encoding_header, params, disable_compression):
     """Bake output for metrics output."""

tests/test_exposition.py

@@ -225,6 +225,7 @@ class TestPushGateway(unittest.TestCase):
     def setUp(self):
         redirect_flag = 'testFlag'
         self.redirect_flag = redirect_flag  # preserve a copy for downstream test assertions
+        self.redirect_status = redirect_status = {'code': 301}
         self.registry = CollectorRegistry()
         self.counter = Gauge('g', 'help', registry=self.registry)
         self.requests = requests = []
@@ -236,7 +237,7 @@ def do_PUT(self):
                 elif 'redirect' in self.requestline and redirect_flag not in self.requestline:
                     # checks for an initial test request with 'redirect' but without the redirect_flag,
                     # and simulates a redirect to a url with the redirect_flag (which will produce a 201)
-                    self.send_response(301)
+                    self.send_response(redirect_status['code'])
                     self.send_header('Location', getattr(self, 'redirect_address', None))
                 else:
                     self.send_response(201)
@@ -408,6 +409,27 @@ def my_redirect_handler(url, method, timeout, headers, data):
         # ensure the redirect took place at the expected redirect location.
         self.assertEqual(self.requests[1][0].path, "/" + self.redirect_flag)
 
+    def _assert_redirect_forwarded(self, command, body):
+        self.assertEqual(len(self.requests), 2)
+        self.assertEqual(self.requests[0][0].command, command)
+        self.assertEqual(self.requests[1][0].command, command)
+        self.assertEqual(self.requests[1][0].path, "/" + self.redirect_flag)
+        self.assertEqual(self.requests[1][1], body)
+
+    def test_push_with_redirect_handler_307(self):
+        self.redirect_status['code'] = 307
+        push_to_gateway(self.address, "my_job_with_redirect", self.registry, handler=passthrough_redirect_handler)
+        self._assert_redirect_forwarded('PUT', b'# HELP g help\n# TYPE g gauge\ng 0.0\n')
+
+    def test_pushadd_with_redirect_handler_308(self):
+        self.redirect_status['code'] = 308
+        pushadd_to_gateway(self.address, "my_job_with_redirect", self.registry, handler=passthrough_redirect_handler)
+        self._assert_redirect_forwarded('POST', b'# HELP g help\n# TYPE g gauge\ng 0.0\n')
+
+    def test_delete_with_redirect_handler(self):
+        delete_from_gateway(self.address, "my_job_with_redirect", handler=passthrough_redirect_handler)
+        self._assert_redirect_forwarded('DELETE', b'')
+
     def test_push_with_trailing_slash(self):
         address = self.address + '/'
         push_to_gateway(address, "my_job_with_trailing_slash", self.registry)