test: request method for specific http code

test/conftest.py

@@ -10,7 +10,7 @@ import subprocess
 import time
 from collections.abc import Callable
 from io import StringIO
-from typing import Iterator, List, Optional, Any
+from typing import Iterator, List, Optional, Any, Iterable
 
 import backoff
 import docker.errors
@@ -108,7 +108,7 @@ class RequestsForDocker:
     def _with_backoff(self, method_name: str, predicate: Callable[[Any], bool], *args, **kwargs) -> Response:
         """Apply backoff retry logic to any session HTTP method."""
         with ipv6(kwargs.pop('ipv6', False)):
-            @backoff.on_predicate(backoff.constant, predicate, interval=.25, max_tries=20)
+            @backoff.on_predicate(backoff.constant, predicate, interval=.25, max_tries=20, jitter = None)
             def _request(*_args, **_kwargs):
                 return getattr(self.session, method_name)(*_args, **_kwargs)
             return _request(*args, **kwargs)
@@ -120,6 +120,16 @@ class RequestsForDocker:
             *args,
             **kwargs
         )
+
+    def get_with_code(self, codes: int | Iterable[int], *args, **kwargs) -> Response:
+        if isinstance(codes, int):
+            codes = [codes]
+        return self._with_backoff(
+            'get',
+            lambda r: r.status_code not in codes,
+            *args,
+            **kwargs
+        )
     
     def get(self, *args, **kwargs) -> Response:
         return self._with_backoff('get', lambda r: r.status_code in (404, 502), *args, **kwargs)

test/test_acme-http-challenge-location/test_acme-http-challenge-location-accept-unknown-host.py

@@ -6,11 +6,12 @@ def test_redirect_acme_challenge_location_enabled(docker_compose, nginxproxy, ac
     assert r.status_code == 200
 
 def test_redirect_acme_challenge_location_disabled(docker_compose, nginxproxy, acme_challenge_path):
-    r = nginxproxy.get(
+    r = nginxproxy.get_with_code(
+        301,
         f"http://web2.nginx-proxy.tld/{acme_challenge_path}",
         allow_redirects=False
     )
-    assert r.status_code == 301
+    assert r.is_permanent_redirect
 
 def test_noredirect_acme_challenge_location_enabled(docker_compose, nginxproxy, acme_challenge_path):
     r = nginxproxy.get(
@@ -20,7 +21,8 @@ def test_noredirect_acme_challenge_location_enabled(docker_compose, nginxproxy,
     assert r.status_code == 200
 
 def test_noredirect_acme_challenge_location_disabled(docker_compose, nginxproxy, acme_challenge_path):
-    r = nginxproxy.get(
+    r = nginxproxy.get_with_code(
+        404,
         f"http://web4.nginx-proxy.tld/{acme_challenge_path}",
         allow_redirects=False
     )

test/test_acme-http-challenge-location/test_acme-http-challenge-location-disabled.py

@@ -1,9 +1,10 @@
 def test_redirect_acme_challenge_location_disabled(docker_compose, nginxproxy, acme_challenge_path):
-    r = nginxproxy.get(
+    r = nginxproxy.get_with_code(
+        301,
         f"http://web1.nginx-proxy.tld/{acme_challenge_path}",
         allow_redirects=False
     )
-    assert r.status_code == 301
+    assert r.is_permanent_redirect
 
 def test_redirect_acme_challenge_location_enabled(docker_compose, nginxproxy, acme_challenge_path):
     r = nginxproxy.get(
@@ -13,7 +14,8 @@ def test_redirect_acme_challenge_location_enabled(docker_compose, nginxproxy, ac
     assert r.status_code == 200
 
 def test_noredirect_acme_challenge_location_disabled(docker_compose, nginxproxy, acme_challenge_path):
-    r = nginxproxy.get(
+    r = nginxproxy.get_with_code(
+        404,
         f"http://web3.nginx-proxy.tld/{acme_challenge_path}",
         allow_redirects=False
     )

test/test_acme-http-challenge-location/test_acme-http-challenge-location-enabled-is-default.py

@@ -6,11 +6,12 @@ def test_redirect_acme_challenge_location_enabled(docker_compose, nginxproxy, ac
     assert r.status_code == 200
 
 def test_redirect_acme_challenge_location_disabled(docker_compose, nginxproxy, acme_challenge_path):
-    r = nginxproxy.get(
+    r = nginxproxy.get_with_code(
+        301,
         f"http://web2.nginx-proxy.tld/{acme_challenge_path}",
         allow_redirects=False
     )
-    assert r.status_code == 301
+    assert r.is_permanent_redirect
 
 def test_noredirect_acme_challenge_location_enabled(docker_compose, nginxproxy, acme_challenge_path):
     r = nginxproxy.get(
@@ -21,6 +22,7 @@ def test_noredirect_acme_challenge_location_enabled(docker_compose, nginxproxy,
 
 def test_noredirect_acme_challenge_location_disabled(docker_compose, nginxproxy, acme_challenge_path):
     r = nginxproxy.get(
+        404,
         f"http://web4.nginx-proxy.tld/{acme_challenge_path}",
         allow_redirects=False
     )

test/test_acme-http-challenge-location/test_acme-http-challenge-location-legacy.py

@@ -6,7 +6,8 @@ def test_redirect_acme_challenge_location_legacy(docker_compose, nginxproxy, acm
     assert r.status_code == 200
 
 def test_noredirect_acme_challenge_location_legacy(docker_compose, nginxproxy, acme_challenge_path):
-    r = nginxproxy.get(
+    r = nginxproxy.get_with_code(
+        404,
         f"http://web2.nginx-proxy.tld/{acme_challenge_path}",
         allow_redirects=False
     )

test/test_custom-error-page/test_custom-error-page.py

@@ -2,6 +2,6 @@ import re
 
 
 def test_custom_error_page(docker_compose, nginxproxy):
-    r = nginxproxy.get("http://unknown.nginx-proxy.tld")
+    r = nginxproxy.get_with_code(503, "http://unknown.nginx-proxy.tld")
     assert r.status_code == 503
     assert re.search(r"Damn, there's some maintenance in progress.", r.text)

test/test_debug-endpoint/test_global.py

@@ -44,5 +44,5 @@ def test_debug_endpoint_hostname_replaced_by_warning_if_regexp(docker_compose, n
 
 
 def test_debug_endpoint_is_disabled_per_container(docker_compose, nginxproxy):
-    r = nginxproxy.get("http://disabled.debug.nginx-proxy.example/nginx-proxy-debug")
+    r = nginxproxy.get_with_code(404, "http://disabled.debug.nginx-proxy.example/nginx-proxy-debug")
     assert r.status_code == 404  

test/test_debug-endpoint/test_per-container.py

@@ -4,9 +4,9 @@ import pytest
 
 
 def test_debug_endpoint_is_disabled_globally(docker_compose, nginxproxy):
-    r = nginxproxy.get("http://disabled1.debug.nginx-proxy.example/nginx-proxy-debug")
+    r = nginxproxy.get_with_code(404,"http://disabled1.debug.nginx-proxy.example/nginx-proxy-debug")
     assert r.status_code == 404 
-    r = nginxproxy.get("http://disabled2.debug.nginx-proxy.example/nginx-proxy-debug")
+    r = nginxproxy.get_with_code(404,"http://disabled2.debug.nginx-proxy.example/nginx-proxy-debug")
     assert r.status_code == 404 
 
 

test/test_enable-http-on-missing-cert/test_enable-http-on-missing-cert.py

@@ -7,8 +7,11 @@ def test_nohttp_missing_cert_enabled(docker_compose, nginxproxy):
     assert r.status_code == 200
 
 def test_redirect_missing_cert_disabled(docker_compose, nginxproxy):
-    r = nginxproxy.get("http://redirect-missing-cert-disabled.nginx-proxy.tld/", allow_redirects=False)
-    assert r.status_code == 301
+    r = nginxproxy.get_with_code(
+        301,
+        "http://redirect-missing-cert-disabled.nginx-proxy.tld/",
+        allow_redirects=False)
+    assert r.is_permanent_redirect
 
 def test_redirect_missing_cert_enabled(docker_compose, nginxproxy):
     r = nginxproxy.get("http://redirect-missing-cert-enabled.nginx-proxy.tld/", allow_redirects=False)

test/test_events/test_events.py

@@ -74,7 +74,7 @@ def test_new_container_is_detected_vpath(web2, nginxproxy):
     r = nginxproxy.get("http://nginx-proxy/web2/port")
     assert r.status_code == 200
     assert "answer from port 82\n" == r.text
-    r = nginxproxy.get("http://nginx-proxy/port")
+    r = nginxproxy.get_with_code([404, 503],"http://nginx-proxy/port")
     assert r.status_code in [404, 503]
 
     web2.remove(force=True)

test/test_htpasswd/test_htpasswd-regex-virtual-host.py

@@ -1,5 +1,5 @@
 def test_htpasswd_regex_virtual_host_is_restricted(docker_compose, nginxproxy):
-    r = nginxproxy.get("http://regex.htpasswd.nginx-proxy.example/port")
+    r = nginxproxy.get_with_code(401, "http://regex.htpasswd.nginx-proxy.example/port")
     assert r.status_code == 401
     assert "WWW-Authenticate" in r.headers
     assert r.headers["WWW-Authenticate"] == 'Basic realm="Restricted access"'

test/test_htpasswd/test_htpasswd-virtual-host.py

@@ -1,5 +1,5 @@
 def test_htpasswd_virtual_host_is_restricted(docker_compose, nginxproxy):
-    r = nginxproxy.get("http://htpasswd.nginx-proxy.tld/port")
+    r = nginxproxy.get_with_code(401, "http://htpasswd.nginx-proxy.tld/port")
     assert r.status_code == 401
     assert "WWW-Authenticate" in r.headers
     assert r.headers["WWW-Authenticate"] == 'Basic realm="Restricted htpasswd.nginx-proxy.tld"'

test/test_htpasswd/test_htpasswd-virtual-path.py

@@ -1,5 +1,5 @@
 def test_htpasswd_virtual_path_is_restricted(docker_compose, nginxproxy):
-    r = nginxproxy.get("http://htpasswd.nginx-proxy.tld/foo/port")
+    r = nginxproxy.get_with_code(401, "http://htpasswd.nginx-proxy.tld/foo/port")
     assert r.status_code == 401
     assert "WWW-Authenticate" in r.headers
     assert r.headers["WWW-Authenticate"] == 'Basic realm="Restricted htpasswd.nginx-proxy.tld/foo/"'

test/test_location-override/test_location-override.py

@@ -1,19 +1,19 @@
 def test_explicit_root_nohash(docker_compose, nginxproxy):
-    r = nginxproxy.get("http://explicit-root-nohash.nginx-proxy.test/port")
+    r = nginxproxy.get_with_code(418, "http://explicit-root-nohash.nginx-proxy.test/port")
     assert r.status_code == 418
     r = nginxproxy.get("http://explicit-root-nohash.nginx-proxy.test/foo/port")
     assert r.status_code == 200
     assert r.text == "answer from port 82\n"
 
 def test_explicit_root_hash(docker_compose, nginxproxy):
-    r = nginxproxy.get("http://explicit-root-hash.nginx-proxy.test/port")
+    r = nginxproxy.get_with_code(418, "http://explicit-root-hash.nginx-proxy.test/port")
     assert r.status_code == 418
     r = nginxproxy.get("http://explicit-root-hash.nginx-proxy.test/foo/port")
     assert r.status_code == 200
     assert r.text == "answer from port 82\n"
 
 def test_explicit_root_hash_and_nohash(docker_compose, nginxproxy):
-    r = nginxproxy.get("http://explicit-root-hash-and-nohash.nginx-proxy.test/port")
+    r = nginxproxy.get_with_code(418, "http://explicit-root-hash-and-nohash.nginx-proxy.test/port")
     assert r.status_code == 418
     r = nginxproxy.get("http://explicit-root-hash-and-nohash.nginx-proxy.test/foo/port")
     assert r.status_code == 200
@@ -23,17 +23,17 @@ def test_explicit_nonroot(docker_compose, nginxproxy):
     r = nginxproxy.get("http://explicit-nonroot.nginx-proxy.test/port")
     assert r.status_code == 200
     assert r.text == "answer from port 81\n"
-    r = nginxproxy.get("http://explicit-nonroot.nginx-proxy.test/foo/port")
+    r = nginxproxy.get_with_code(418, "http://explicit-nonroot.nginx-proxy.test/foo/port")
     assert r.status_code == 418
 
 def test_implicit_root_nohash(docker_compose, nginxproxy):
-    r = nginxproxy.get("http://implicit-root-nohash.nginx-proxy.test/port")
+    r = nginxproxy.get_with_code(418, "http://implicit-root-nohash.nginx-proxy.test/port")
     assert r.status_code == 418
 
 def test_implicit_root_hash(docker_compose, nginxproxy):
-    r = nginxproxy.get("http://implicit-root-hash.nginx-proxy.test/port")
+    r = nginxproxy.get_with_code(418, "http://implicit-root-hash.nginx-proxy.test/port")
     assert r.status_code == 418
 
 def test_implicit_root_hash_and_nohash(docker_compose, nginxproxy):
-    r = nginxproxy.get("http://implicit-root-hash-and-nohash.nginx-proxy.test/port")
+    r = nginxproxy.get_with_code(418, "http://implicit-root-hash-and-nohash.nginx-proxy.test/port")
     assert r.status_code == 418

test/test_ports/test_virtual-port-single-different-from-single-port.py

@@ -2,6 +2,6 @@ import re
 
 
 def test_answer_is_served_from_virtual_port_which_is_ureachable(docker_compose, nginxproxy):
-    r = nginxproxy.get("http://web.nginx-proxy.tld/port")
+    r = nginxproxy.get_with_code(502, "http://web.nginx-proxy.tld/port")
     assert r.status_code == 502
     assert re.search(r"\n\s+server \d+\.\d+\.\d+\.\d+:90;\n", nginxproxy.get_conf().decode('ASCII'))

test/test_server-down/test_server-down.py

@@ -1,5 +1,5 @@
 def test_web_has_server_down(docker_compose, nginxproxy):
     conf = nginxproxy.get_conf().decode('ASCII')
-    r = nginxproxy.get("http://web.nginx-proxy.tld/port")
+    r = nginxproxy.get_with_code([502, 503], "http://web.nginx-proxy.tld/port")
     assert r.status_code in [502, 503]
     assert conf.count("server 127.0.0.1 down;") == 1

test/test_ssl/test_https-port.py

@@ -3,8 +3,8 @@ import pytest
 
 @pytest.mark.parametrize("subdomain", ["foo", "bar"])
 def test_web1_http_redirects_to_https(docker_compose, nginxproxy, subdomain):
-    r = nginxproxy.get("http://%s.nginx-proxy.tld:8080/" % subdomain, allow_redirects=False)
-    assert r.status_code == 301
+    r = nginxproxy.get_with_code(301, "http://%s.nginx-proxy.tld:8080/" % subdomain, allow_redirects=False)
+    assert r.is_permanent_redirect
     assert "Location" in r.headers
     assert "https://%s.nginx-proxy.tld:8443/" % subdomain == r.headers['Location']
 

test/test_ssl/test_nohttps.py

@@ -14,7 +14,8 @@ def test_https_is_disabled(docker_compose, nginxproxy):
 
 
 def test_http_acme_challenge_does_not_work(docker_compose, nginxproxy, acme_challenge_path):
-    r = nginxproxy.get(
+    r = nginxproxy.get_with_code(
+        404,
         f"http://web.nginx-proxy.tld/{acme_challenge_path}",
         allow_redirects=False
     )

test/test_ssl/test_virtual-path.py

@@ -4,8 +4,8 @@ from requests import ConnectionError
 
 @pytest.mark.parametrize("path", ["web1", "web2"])
 def test_web1_http_redirects_to_https(docker_compose, nginxproxy, path):
-    r = nginxproxy.get("http://www.nginx-proxy.tld/%s/port" % path, allow_redirects=False)
-    assert r.status_code == 301
+    r = nginxproxy.get_with_code(301, "http://www.nginx-proxy.tld/%s/port" % path, allow_redirects=False)
+    assert r.is_permanent_redirect
     assert "Location" in r.headers
     assert "https://www.nginx-proxy.tld/%s/port" % path == r.headers['Location']
 

test/test_ssl/test_wildcard-cert-nohttps.py

@@ -44,7 +44,8 @@ def test_acme_challenge_works(
     docker_compose, nginxproxy, acme_challenge_path, subdomain, acme_should_work
 ):
     if acme_should_work:
-        r = nginxproxy.get(
+        r = nginxproxy.get_with_code(
+            404,
             f"https://{subdomain}.web.nginx-proxy.tld/{acme_challenge_path}",
             allow_redirects=False
         )

test/test_ssl/test_wildcard.py

@@ -3,8 +3,8 @@ import pytest
 
 @pytest.mark.parametrize("subdomain", ["foo", "bar"])
 def test_web1_http_redirects_to_https(docker_compose, nginxproxy, subdomain):
-    r = nginxproxy.get(f"http://{subdomain}.nginx-proxy.tld/", allow_redirects=False)
-    assert r.status_code == 301
+    r = nginxproxy.get_with_code(301, f"http://{subdomain}.nginx-proxy.tld/", allow_redirects=False)
+    assert r.is_permanent_redirect
     assert "Location" in r.headers
     assert f"https://{subdomain}.nginx-proxy.tld/" == r.headers['Location']
 

test/test_virtual-path/test_custom-conf.py

@@ -2,7 +2,7 @@ import pytest
 
 
 def test_default_root_response(docker_compose, nginxproxy):
-    r = nginxproxy.get("http://nginx-proxy.test/")
+    r = nginxproxy.get_with_code(418, "http://nginx-proxy.test/")
     assert r.status_code == 418
 
 @pytest.mark.parametrize("stub,header", [
@@ -22,7 +22,7 @@ def test_custom_applies(docker_compose, nginxproxy, stub, header):
     ("bar.nginx-proxy.test", 503),
 ])
 def test_custom_does_not_apply(docker_compose, nginxproxy, stub, code):
-    r = nginxproxy.get(f"http://{stub}/port")
+    r = nginxproxy.get_with_code(code, f"http://{stub}/port")
     assert r.status_code == code
     assert "X-test" not in r.headers
 

test/test_virtual-path/test_forwarding.py

@@ -1,6 +1,6 @@
 def test_root_redirects_to_web1(docker_compose, nginxproxy):
-    r = nginxproxy.get("http://www.nginx-proxy.tld/port", allow_redirects=False)
-    assert r.status_code == 301
+    r = nginxproxy.get_with_code(301, "http://www.nginx-proxy.tld/port", allow_redirects=False)
+    assert r.is_permanent_redirect
     assert "Location" in r.headers
     assert "http://www.nginx-proxy.tld/web1/port" == r.headers['Location']
 

test/test_virtual-path/test_virtual-paths.py

@@ -20,7 +20,7 @@ def test_valid_path(docker_compose, nginxproxy, stub, expected_port):
     "bar.nginx-proxy.test",
 ])
 def test_invalid_path(docker_compose, nginxproxy, stub):
-    r = nginxproxy.get(f"http://{stub}/port")
+    r = nginxproxy.get_with_code([404, 503], f"http://{stub}/port")
     assert r.status_code in [404, 503]
 
 @pytest.fixture
@@ -57,5 +57,5 @@ def test_container_hotplug(web4, nginxproxy):
     assert r.text == f"answer from port 84\n"
     web4.remove(force=True)
     sleep(2)
-    r = nginxproxy.get(f"http://nginx-proxy.test/web4/port")
+    r = nginxproxy.get_with_code(404, f"http://nginx-proxy.test/web4/port")
     assert r.status_code == 404