<gh_stars>10-100
import io
import json
import pytest
import time
from ahip.base import DEFAULT_PORTS
from ahip import PoolManager, Retry
from ahip.exceptions import MaxRetryError, NewConnectionError, UnrewindableBodyError
from hip.util.retry import RequestHistory

from dummyserver.server import HAS_IPV6
from dummyserver.testcase import HTTPDummyServerTestCase, IPv6HTTPDummyServerTestCase

from test import LONG_TIMEOUT
from test.with_dummyserver import conftest


def _test_name(test_name, backend, anyio_backend):
    return "-".join([test_name, backend, anyio_backend or "none"])


class TestPoolManager(HTTPDummyServerTestCase):
    @classmethod
    def setup_class(cls):
        super(TestPoolManager, cls).setup_class()
        cls.base_url = "http://%s:%d" % (cls.host, cls.port)
        cls.base_url_alt = "http://%s:%d" % (cls.host_alt, cls.port)

    @conftest.test_all_backends
    async def test_redirect(self, backend, anyio_backend):
        with PoolManager(backend=backend) as http:
            r = await http.request(
                "GET",
                "%s/redirect" % self.base_url,
                fields={"target": "%s/" % self.base_url},
                redirect=False,
            )
            assert r.status == 303

            r = await http.request(
                "GET",
                "%s/redirect" % self.base_url,
                fields={"target": "%s/" % self.base_url},
            )

            assert r.status == 200
            assert r.data == b"Dummy server!"

    @conftest.test_all_backends
    async def test_redirect_twice(self, backend, anyio_backend):
        with PoolManager(backend=backend) as http:
            r = await http.request(
                "GET",
                "%s/redirect" % self.base_url,
                fields={"target": "%s/redirect" % self.base_url},
                redirect=False,
            )

            assert r.status == 303

            r = await http.request(
                "GET",
                "%s/redirect" % self.base_url,
                fields={
                    "target": "%s/redirect?target=%s/" % (self.base_url, self.base_url)
                },
            )

            assert r.status == 200
            assert r.data == b"Dummy server!"

    @conftest.test_all_backends
    async def test_redirect_to_relative_url(self, backend, anyio_backend):
        with PoolManager(backend=backend) as http:
            r = await http.request(
                "GET",
                "%s/redirect" % self.base_url,
                fields={"target": "/redirect"},
                redirect=False,
            )

            assert r.status == 303

            r = await http.request(
                "GET", "%s/redirect" % self.base_url, fields={"target": "/redirect"}
            )

            assert r.status == 200
            assert r.data == b"Dummy server!"

    @conftest.test_all_backends
    async def test_cross_host_redirect(self, backend, anyio_backend):
        with PoolManager(backend=backend) as http:
            cross_host_location = "%s/echo?a=b" % self.base_url_alt
            with pytest.raises(MaxRetryError):
                await http.request(
                    "GET",
                    "%s/redirect" % self.base_url,
                    fields={"target": cross_host_location},
                    timeout=LONG_TIMEOUT,
                    retries=0,
                )

            r = await http.request(
                "GET",
                "%s/redirect" % self.base_url,
                fields={"target": "%s/echo?a=b" % self.base_url_alt},
                timeout=LONG_TIMEOUT,
                retries=1,
            )

            assert r._pool.host == self.host_alt

    @conftest.test_all_backends
    async def test_too_many_redirects(self, backend, anyio_backend):
        with PoolManager(backend=backend) as http:
            with pytest.raises(MaxRetryError):
                await http.request(
                    "GET",
                    "%s/redirect" % self.base_url,
                    fields={
                        "target": "%s/redirect?target=%s/"
                        % (self.base_url, self.base_url)
                    },
                    retries=1,
                )

            with pytest.raises(MaxRetryError):
                await http.request(
                    "GET",
                    "%s/redirect" % self.base_url,
                    fields={
                        "target": "%s/redirect?target=%s/"
                        % (self.base_url, self.base_url)
                    },
                    retries=Retry(total=None, redirect=1),
                )

    @conftest.test_all_backends
    async def test_redirect_cross_host_remove_headers(self, backend, anyio_backend):
        with PoolManager(backend=backend) as http:
            r = await http.request(
                "GET",
                "%s/redirect" % self.base_url,
                fields={"target": "%s/headers" % self.base_url_alt},
                headers={"Authorization": "foo"},
            )

            assert r.status == 200

            data = json.loads(r.data.decode("utf-8"))

            assert "Authorization" not in data

            r = await http.request(
                "GET",
                "%s/redirect" % self.base_url,
                fields={"target": "%s/headers" % self.base_url_alt},
                headers={"authorization": "foo"},
            )

            assert r.status == 200

            data = json.loads(r.data.decode("utf-8"))

            assert "authorization" not in data
            assert "Authorization" not in data

    @conftest.test_all_backends
    async def test_redirect_cross_host_no_remove_headers(self, backend, anyio_backend):
        with PoolManager(backend=backend) as http:
            r = await http.request(
                "GET",
                "%s/redirect" % self.base_url,
                fields={"target": "%s/headers" % self.base_url_alt},
                headers={"Authorization": "foo"},
                retries=Retry(remove_headers_on_redirect=[]),
            )

            assert r.status == 200

            data = json.loads(r.data.decode("utf-8"))

            assert data["Authorization"] == "foo"

    @conftest.test_all_backends
    async def test_redirect_cross_host_set_removed_headers(
        self, backend, anyio_backend
    ):
        with PoolManager(backend=backend) as http:
            r = await http.request(
                "GET",
                "%s/redirect" % self.base_url,
                fields={"target": "%s/headers" % self.base_url_alt},
                headers={"X-API-Secret": "foo", "Authorization": "bar"},
                retries=Retry(remove_headers_on_redirect=["X-API-Secret"]),
            )

            assert r.status == 200

            data = json.loads(r.data.decode("utf-8"))

            assert "X-API-Secret" not in data
            assert data["Authorization"] == "bar"

            r = await http.request(
                "GET",
                "%s/redirect" % self.base_url,
                fields={"target": "%s/headers" % self.base_url_alt},
                headers={"x-api-secret": "foo", "authorization": "bar"},
                retries=Retry(remove_headers_on_redirect=["X-API-Secret"]),
            )

            assert r.status == 200

            data = json.loads(r.data.decode("utf-8"))

            assert "x-api-secret" not in data
            assert "X-API-Secret" not in data
            assert data["Authorization"] == "bar"

    @conftest.test_all_backends
    async def test_raise_on_redirect(self, backend, anyio_backend):
        with PoolManager(backend=backend) as http:
            r = await http.request(
                "GET",
                "%s/redirect" % self.base_url,
                fields={
                    "target": "%s/redirect?target=%s/" % (self.base_url, self.base_url)
                },
                retries=Retry(total=None, redirect=1, raise_on_redirect=False),
            )

            assert r.status == 303

    @conftest.test_all_backends
    async def test_raise_on_status(self, backend, anyio_backend):
        with PoolManager(backend=backend) as http:
            with pytest.raises(MaxRetryError):
                # the default is to raise
                r = await http.request(
                    "GET",
                    "%s/status" % self.base_url,
                    fields={"status": "500 Internal Server Error"},
                    retries=Retry(total=1, status_forcelist=range(500, 600)),
                )

            with pytest.raises(MaxRetryError):
                # raise explicitly
                r = await http.request(
                    "GET",
                    "%s/status" % self.base_url,
                    fields={"status": "500 Internal Server Error"},
                    retries=Retry(
                        total=1, status_forcelist=range(500, 600), raise_on_status=True
                    ),
                )

            # don't raise
            r = await http.request(
                "GET",
                "%s/status" % self.base_url,
                fields={"status": "500 Internal Server Error"},
                retries=Retry(
                    total=1, status_forcelist=range(500, 600), raise_on_status=False
                ),
            )

            assert r.status == 500

    @pytest.mark.parametrize(
        ["target", "expected_target"],
        [
            ("/echo_uri?q=1#fragment", b"/echo_uri?q=1"),
            ("/echo_uri?#", b"/echo_uri?"),
            ("/echo_uri#?", b"/echo_uri"),
            ("/echo_uri#?#", b"/echo_uri"),
            ("/echo_uri??#", b"/echo_uri??"),
            ("/echo_uri?%3f#", b"/echo_uri?%3F"),
            ("/echo_uri?%3F#", b"/echo_uri?%3F"),
            ("/echo_uri?[]", b"/echo_uri?%5B%5D"),
        ],
    )
    @conftest.test_all_backends
    async def test_encode_http_target(
        self, target, expected_target, backend, anyio_backend
    ):
        with PoolManager() as http:
            url = "http://%s:%d%s" % (self.host, self.port, target)
            r = await http.request("GET", url)
            assert r.data == expected_target

    @conftest.test_all_backends
    async def test_missing_port(self, backend, anyio_backend):
        # Can a URL that lacks an explicit port like ':80' succeed, or
        # will all such URLs fail with an error?

        with PoolManager(backend=backend) as http:
            # By globally adjusting `DEFAULT_PORTS` we pretend for a moment
            # that HTTP's default port is not 80, but is the port at which
            # our test server happens to be listening.
            DEFAULT_PORTS["http"] = self.port
            try:
                r = await http.request("GET", "http://%s/" % self.host, retries=0)
            finally:
                DEFAULT_PORTS["http"] = 80

            assert r.status == 200
            assert r.data == b"Dummy server!"

    @conftest.test_all_backends
    async def test_headers(self, backend, anyio_backend):
        with PoolManager(backend=backend, headers={"Foo": "bar"}) as http:
            r = await http.request("GET", "%s/headers" % self.base_url)
            returned_headers = json.loads(r.data.decode())
            assert returned_headers.get("Foo") == "bar"

            r = await http.request("POST", "%s/headers" % self.base_url)
            returned_headers = json.loads(r.data.decode())
            assert returned_headers.get("Foo") == "bar"

            r = await http.request_encode_url("GET", "%s/headers" % self.base_url)
            returned_headers = json.loads(r.data.decode())
            assert returned_headers.get("Foo") == "bar"

            r = await http.request_encode_body("POST", "%s/headers" % self.base_url)
            returned_headers = json.loads(r.data.decode())
            assert returned_headers.get("Foo") == "bar"

            r = await http.request_encode_url(
                "GET", "%s/headers" % self.base_url, headers={"Baz": "quux"}
            )
            returned_headers = json.loads(r.data.decode())
            assert returned_headers.get("Foo") is None
            assert returned_headers.get("Baz") == "quux"

            r = await http.request_encode_body(
                "GET", "%s/headers" % self.base_url, headers={"Baz": "quux"}
            )
            returned_headers = json.loads(r.data.decode())
            assert returned_headers.get("Foo") is None
            assert returned_headers.get("Baz") == "quux"

    @conftest.test_all_backends
    async def test_http_with_ssl_keywords(self, backend, anyio_backend):
        with PoolManager(backend=backend, ca_certs="REQUIRED") as http:
            r = await http.request("GET", "http://%s:%s/" % (self.host, self.port))
            assert r.status == 200

    @conftest.test_all_backends
    async def test_http_with_ca_cert_dir(self, backend, anyio_backend):
        with PoolManager(
            backend=backend, ca_certs="REQUIRED", ca_cert_dir="/nosuchdir"
        ) as http:
            r = await http.request("GET", "http://%s:%s/" % (self.host, self.port))
            assert r.status == 200

    @conftest.test_all_backends
    async def test_cleanup_on_connection_error(self, backend, anyio_backend):
        """
        Test that connections are recycled to the pool on
        connection errors where no http response is received.
        """
        poolsize = 3

        with PoolManager(backend=backend, maxsize=poolsize, block=True) as http:
            pool = http.connection_from_host(self.host, self.port)
            assert pool.pool.qsize() == poolsize

            # force a connection error by supplying a non-existent
            # url. We won't get a response for this  and so the
            # conn won't be implicitly returned to the pool.
            url = "%s/redirect" % self.base_url
            with pytest.raises(MaxRetryError):
                await http.request("GET", url, fields={"target": "/"}, retries=0)

            r = await http.request("GET", url, fields={"target": "/"}, retries=1)
            r.release_conn()

            # the pool should still contain poolsize elements
            assert pool.pool.qsize() == poolsize


class TestFileUploads(HTTPDummyServerTestCase):
    @classmethod
    def setup_class(cls):
        super(TestFileUploads, cls).setup_class()
        cls.base_url = "http://%s:%d" % (cls.host, cls.port)
        cls.base_url_alt = "http://%s:%d" % (cls.host_alt, cls.port)

    @conftest.test_all_backends
    async def test_redirect_put_file(self, backend, anyio_backend):
        """PUT with file object should work with a redirection response"""
        retry = Retry(total=3, status_forcelist=[418])
        # httplib reads in 8k chunks; use a larger content length
        content_length = 65535
        data = b"A" * content_length
        uploaded_file = io.BytesIO(data)
        headers = {
            "test-name": "test_redirect_put_file",
            "Content-Length": str(content_length),
        }
        url = "%s/redirect?target=/echo&status=307" % self.base_url

        with PoolManager(backend=backend) as http:
            resp = await http.urlopen(
                "PUT", url, headers=headers, retries=retry, body=uploaded_file
            )
            assert resp.status == 200
            assert resp.data == data

    @conftest.test_all_backends
    async def test_retries_put_filehandle(self, backend, anyio_backend):
        """HTTP PUT retry with a file-like object should not timeout"""
        retry = Retry(total=3, status_forcelist=[418])
        # httplib reads in 8k chunks; use a larger content length
        content_length = 65535
        data = b"A" * content_length
        uploaded_file = io.BytesIO(data)
        headers = {
            "test-name": _test_name(
                "test_retries_put_filehandle", backend, anyio_backend
            ),
            "Content-Length": str(content_length),
        }

        with PoolManager(backend=backend) as http:
            resp = await http.urlopen(
                "PUT",
                "%s/successful_retry" % self.base_url,
                headers=headers,
                retries=retry,
                body=uploaded_file,
                redirect=False,
            )
            assert resp.status == 200

    @conftest.test_all_backends
    async def test_redirect_with_failed_tell(self, backend, anyio_backend):
        """Abort request if failed to get a position from tell()"""

        class BadTellObject(io.BytesIO):
            def tell(self):
                raise IOError

        body = BadTellObject(b"the data")
        url = "%s/redirect?target=/successful_retry" % self.base_url
        # httplib uses fileno if Content-Length isn't supplied,
        # which is unsupported by BytesIO.
        headers = {"Content-Length": "8"}

        with PoolManager(backend=backend) as http:
            with pytest.raises(UnrewindableBodyError) as e:
                await http.urlopen("PUT", url, headers=headers, body=body)
            assert "Unable to record file position for" in str(e.value)

    @conftest.test_all_backends
    async def test_redirect_with_failed_seek(self, backend, anyio_backend):
        """Abort request if failed to restore position with seek()"""

        class BadSeekObject(io.BytesIO):
            def seek(self, *_):
                raise IOError

        body = BadSeekObject(b"the data")
        url = "%s/redirect?target=/successful_retry" % self.base_url
        # httplib uses fileno if Content-Length isn't supplied,
        # which is unsupported by BytesIO.
        headers = {"Content-Length": "8"}

        with PoolManager(backend=backend) as http:
            with pytest.raises(UnrewindableBodyError) as e:
                await http.urlopen("PUT", url, headers=headers, body=body)
            assert (
                "An error occurred when rewinding request body for redirect/retry."
                == str(e.value)
            )


class TestRetry(HTTPDummyServerTestCase):
    @classmethod
    def setup_class(self):
        super(TestRetry, self).setup_class()
        self.base_url = "http://%s:%d" % (self.host, self.port)
        self.base_url_alt = "http://%s:%d" % (self.host_alt, self.port)

    @conftest.test_all_backends
    async def test_max_retry(self, backend, anyio_backend):
        with PoolManager(backend=backend) as http:
            with pytest.raises(MaxRetryError):
                await http.request(
                    "GET",
                    "%s/redirect" % self.base_url,
                    fields={"target": "/"},
                    retries=0,
                )

    @conftest.test_all_backends
    async def test_disabled_retry(self, backend, anyio_backend):
        """ Disabled retries should disable redirect handling. """
        with PoolManager(backend=backend) as http:
            r = await http.request(
                "GET",
                "%s/redirect" % self.base_url,
                fields={"target": "/"},
                retries=False,
            )
            assert r.status == 303

            r = await http.request(
                "GET",
                "%s/redirect" % self.base_url,
                fields={"target": "/"},
                retries=Retry(redirect=False),
            )
            assert r.status == 303

            with pytest.raises(NewConnectionError):
                await http.request(
                    "GET",
                    "http://thishostdoesnotexist.invalid/",
                    timeout=0.001,
                    retries=False,
                )

    @conftest.test_all_backends
    async def test_read_retries(self, backend, anyio_backend):
        """ Should retry for status codes in the whitelist """
        retry = Retry(read=1, status_forcelist=[418])

        with PoolManager(backend=backend) as http:
            resp = await http.request(
                "GET",
                "%s/successful_retry" % self.base_url,
                headers={
                    "test-name": _test_name("test_read_retries", backend, anyio_backend)
                },
                retries=retry,
            )
            assert resp.status == 200

    @conftest.test_all_backends
    async def test_read_total_retries(self, backend, anyio_backend):
        """ HTTP response w/ status code in the whitelist should be retried """
        headers = {
            "test-name": _test_name("test_read_total_retries", backend, anyio_backend)
        }
        retry = Retry(total=1, status_forcelist=[418])

        with PoolManager(backend=backend) as http:
            resp = await http.request(
                "GET",
                "%s/successful_retry" % self.base_url,
                headers=headers,
                retries=retry,
            )
            assert resp.status == 200

    @conftest.test_all_backends
    async def test_retries_wrong_whitelist(self, backend, anyio_backend):
        """HTTP response w/ status code not in whitelist shouldn't be retried"""
        headers = {
            "test-name": _test_name("test_wrong_whitelist", backend, anyio_backend)
        }
        retry = Retry(total=1, status_forcelist=[202])

        with PoolManager(backend=backend) as http:
            resp = await http.request(
                "GET",
                "%s/successful_retry" % self.base_url,
                retries=retry,
                headers=headers,
            )
            assert resp.status == 418

    @conftest.test_all_backends
    async def test_default_method_whitelist_retried(self, backend, anyio_backend):
        """Hip should retry methods in the default method whitelist"""
        headers = {
            "test-name": _test_name("test_default_whitelist", backend, anyio_backend)
        }
        retry = Retry(total=1, status_forcelist=[418])

        with PoolManager(backend=backend) as http:
            resp = await http.request(
                "OPTIONS",
                "%s/successful_retry" % self.base_url,
                headers=headers,
                retries=retry,
            )
            assert resp.status == 200

    @conftest.test_all_backends
    async def test_retries_wrong_method_list(self, backend, anyio_backend):
        """Method not in our whitelist should not be retried, even if code matches"""
        headers = {
            "test-name": _test_name(
                "test_wrong_method_whitelist", backend, anyio_backend
            )
        }
        retry = Retry(total=1, status_forcelist=[418], method_whitelist=["POST"])

        with PoolManager(backend=backend) as http:
            resp = await http.request(
                "GET",
                "%s/successful_retry" % self.base_url,
                headers=headers,
                retries=retry,
            )
            assert resp.status == 418

    @conftest.test_all_backends
    async def test_read_retries_unsuccessful(self, backend, anyio_backend):
        headers = {
            "test-name": _test_name(
                "test_read_retries_unsuccessful", backend, anyio_backend
            )
        }

        with PoolManager(backend=backend) as http:
            resp = await http.request(
                "GET", "%s/successful_retry" % self.base_url, headers=headers, retries=1
            )
            assert resp.status == 418

    @conftest.test_all_backends
    async def test_retry_reuse_safe(self, backend, anyio_backend):
        """ It should be possible to reuse a Retry object across requests """
        headers = {"test-name": _test_name("test_retry_safe", backend, anyio_backend)}
        retry = Retry(total=1, status_forcelist=[418])

        with PoolManager(backend=backend) as http:
            resp = await http.request(
                "GET",
                "%s/successful_retry" % self.base_url,
                headers=headers,
                retries=retry,
            )
            assert resp.status == 200
            resp = await http.request(
                "GET",
                "%s/successful_retry" % self.base_url,
                headers=headers,
                retries=retry,
            )
            assert resp.status == 200

    @conftest.test_all_backends
    async def test_retry_return_in_response(self, backend, anyio_backend):
        headers = {
            "test-name": _test_name(
                "test_retry_return_in_response", backend, anyio_backend
            )
        }
        retry = Retry(total=2, status_forcelist=[418])

        with PoolManager(backend=backend) as http:
            resp = await http.request(
                "GET",
                "%s/successful_retry" % self.base_url,
                headers=headers,
                retries=retry,
            )
            assert resp.status == 200
            assert resp.retries.total == 1
            assert resp.retries.history == (
                RequestHistory("GET", "/successful_retry", None, 418, None),
            )

    @conftest.test_all_backends
    async def test_retry_redirect_history(self, backend, anyio_backend):
        with PoolManager(backend=backend) as http:
            resp = await http.request(
                "GET", "%s/redirect" % self.base_url, fields={"target": "/"}
            )
            assert resp.status == 200
            assert resp.retries.history == (
                RequestHistory(
                    "GET", self.base_url + "/redirect?target=%2F", None, 303, "/"
                ),
            )

    @conftest.test_all_backends
    async def test_multi_redirect_history(self, backend, anyio_backend):
        with PoolManager(backend=backend) as http:
            r = await http.request(
                "GET",
                "%s/multi_redirect" % self.base_url,
                fields={"redirect_codes": "303,302,200"},
                redirect=False,
            )
            assert r.status == 303
            assert r.retries.history == tuple()

            r = await http.request(
                "GET",
                "%s/multi_redirect" % self.base_url,
                retries=10,
                fields={"redirect_codes": "303,302,301,307,302,200"},
            )
            assert r.status == 200
            assert r.data == b"Done redirecting"

            expected = [
                (303, "/multi_redirect?redirect_codes=302,301,307,302,200"),
                (302, "/multi_redirect?redirect_codes=301,307,302,200"),
                (301, "/multi_redirect?redirect_codes=307,302,200"),
                (307, "/multi_redirect?redirect_codes=302,200"),
                (302, "/multi_redirect?redirect_codes=200"),
            ]
            actual = [
                (history.status, history.redirect_location)
                for history in r.retries.history
            ]
            assert actual == expected


class TestRetryAfter(HTTPDummyServerTestCase):
    @classmethod
    def setup_class(self):
        super(TestRetryAfter, self).setup_class()
        self.base_url = "http://%s:%d" % (self.host, self.port)
        self.base_url_alt = "http://%s:%d" % (self.host_alt, self.port)

    @conftest.test_all_backends
    async def test_retry_after(self, backend, anyio_backend):
        url = "%s/retry_after" % self.base_url
        with PoolManager(backend=backend) as http:
            # Request twice in a second to get a 429 response.
            r = await http.request(
                "GET", url, fields={"status": "429 Too Many Requests"}, retries=False
            )
            r = await http.request(
                "GET", url, fields={"status": "429 Too Many Requests"}, retries=False
            )
            assert r.status == 429

            r = await http.request(
                "GET", url, fields={"status": "429 Too Many Requests"}, retries=True
            )
            assert r.status == 200

            # Request twice in a second to get a 503 response.
            r = await http.request(
                "GET", url, fields={"status": "503 Service Unavailable"}, retries=False
            )
            r = await http.request(
                "GET", url, fields={"status": "503 Service Unavailable"}, retries=False
            )
            assert r.status == 503

            r = await http.request(
                "GET", url, fields={"status": "503 Service Unavailable"}, retries=True
            )
            assert r.status == 200

            # Ignore Retry-After header on status which is not defined in
            # Retry.RETRY_AFTER_STATUS_CODES.
            r = await http.request(
                "GET", url, fields={"status": "418 I'm a teapot"}, retries=True
            )
            assert r.status == 418

    @conftest.test_all_backends
    async def test_redirect_after(self, backend, anyio_backend):
        with PoolManager(backend=backend) as http:
            r = await http.request(
                "GET", "%s/redirect_after" % self.base_url, retries=False
            )
            assert r.status == 303

            t = time.time()
            r = await http.request("GET", "%s/redirect_after" % self.base_url)
            assert r.status == 200
            delta = time.time() - t
            assert delta >= 1

            t = time.time()
            timestamp = t + 2
            r = await http.request(
                "GET", self.base_url + "/redirect_after?date=" + str(timestamp)
            )
            assert r.status == 200
            delta = time.time() - t
            assert delta >= 1

            # Retry-After is past
            t = time.time()
            timestamp = t - 1
            r = await http.request(
                "GET", self.base_url + "/redirect_after?date=" + str(timestamp)
            )
            delta = time.time() - t
            assert r.status == 200
            assert delta < 1


@pytest.mark.skipif(not HAS_IPV6, reason="IPv6 is not supported on this system")
class TestIPv6PoolManager(IPv6HTTPDummyServerTestCase):
    @classmethod
    def setup_class(cls):
        super(TestIPv6PoolManager, cls).setup_class()
        cls.base_url = "http://[%s]:%d" % (cls.host, cls.port)

    @conftest.test_all_backends
    async def test_ipv6(self, backend, anyio_backend):
        with PoolManager(backend=backend) as http:
            await http.request("GET", self.base_url)

import io
import json
import pytest
import time
from ahip.base import DEFAULT_PORTS
from ahip import PoolManager, Retry
from ahip.exceptions import MaxRetryError, NewConnectionError, UnrewindableBodyError
from hip.util.retry import RequestHistory

from dummyserver.server import HAS_IPV6
from dummyserver.testcase import HTTPDummyServerTestCase, IPv6HTTPDummyServerTestCase

from test import LONG_TIMEOUT
from test.with_dummyserver import conftest


def _test_name(test_name, backend, anyio_backend):
    return "-".join([test_name, backend, anyio_backend or "none"])


class TestPoolManager(HTTPDummyServerTestCase):
    @classmethod
    def setup_class(cls):
        super(TestPoolManager, cls).setup_class()
        cls.base_url = "http://%s:%d" % (cls.host, cls.port)
        cls.base_url_alt = "http://%s:%d" % (cls.host_alt, cls.port)

    @conftest.test_all_backends
    async def test_redirect(self, backend, anyio_backend):
        with PoolManager(backend=backend) as http:
            r = await http.request(
                "GET",
                "%s/redirect" % self.base_url,
                fields={"target": "%s/" % self.base_url},
                redirect=False,
            )
            assert r.status == 303

            r = await http.request(
                "GET",
                "%s/redirect" % self.base_url,
                fields={"target": "%s/" % self.base_url},
            )

            assert r.status == 200
            assert r.data == b"Dummy server!"

    @conftest.test_all_backends
    async def test_redirect_twice(self, backend, anyio_backend):
        with PoolManager(backend=backend) as http:
            r = await http.request(
                "GET",
                "%s/redirect" % self.base_url,
                fields={"target": "%s/redirect" % self.base_url},
                redirect=False,
            )

            assert r.status == 303

            r = await http.request(
                "GET",
                "%s/redirect" % self.base_url,
                fields={
                    "target": "%s/redirect?target=%s/" % (self.base_url, self.base_url)
                },
            )

            assert r.status == 200
            assert r.data == b"Dummy server!"

    @conftest.test_all_backends
    async def test_redirect_to_relative_url(self, backend, anyio_backend):
        with PoolManager(backend=backend) as http:
            r = await http.request(
                "GET",
                "%s/redirect" % self.base_url,
                fields={"target": "/redirect"},
                redirect=False,
            )

            assert r.status == 303

            r = await http.request(
                "GET", "%s/redirect" % self.base_url, fields={"target": "/redirect"}
            )

            assert r.status == 200
            assert r.data == b"Dummy server!"

    @conftest.test_all_backends
    async def test_cross_host_redirect(self, backend, anyio_backend):
        with PoolManager(backend=backend) as http:
            cross_host_location = "%s/echo?a=b" % self.base_url_alt
            with pytest.raises(MaxRetryError):
                await http.request(
                    "GET",
                    "%s/redirect" % self.base_url,
                    fields={"target": cross_host_location},
                    timeout=LONG_TIMEOUT,
                    retries=0,
                )

            r = await http.request(
                "GET",
                "%s/redirect" % self.base_url,
                fields={"target": "%s/echo?a=b" % self.base_url_alt},
                timeout=LONG_TIMEOUT,
                retries=1,
            )

            assert r._pool.host == self.host_alt

    @conftest.test_all_backends
    async def test_too_many_redirects(self, backend, anyio_backend):
        with PoolManager(backend=backend) as http:
            with pytest.raises(MaxRetryError):
                await http.request(
                    "GET",
                    "%s/redirect" % self.base_url,
                    fields={
                        "target": "%s/redirect?target=%s/"
                        % (self.base_url, self.base_url)
                    },
                    retries=1,
                )

            with pytest.raises(MaxRetryError):
                await http.request(
                    "GET",
                    "%s/redirect" % self.base_url,
                    fields={
                        "target": "%s/redirect?target=%s/"
                        % (self.base_url, self.base_url)
                    },
                    retries=Retry(total=None, redirect=1),
                )

    @conftest.test_all_backends
    async def test_redirect_cross_host_remove_headers(self, backend, anyio_backend):
        with PoolManager(backend=backend) as http:
            r = await http.request(
                "GET",
                "%s/redirect" % self.base_url,
                fields={"target": "%s/headers" % self.base_url_alt},
                headers={"Authorization": "foo"},
            )

            assert r.status == 200

            data = json.loads(r.data.decode("utf-8"))

            assert "Authorization" not in data

            r = await http.request(
                "GET",
                "%s/redirect" % self.base_url,
                fields={"target": "%s/headers" % self.base_url_alt},
                headers={"authorization": "foo"},
            )

            assert r.status == 200

            data = json.loads(r.data.decode("utf-8"))

            assert "authorization" not in data
            assert "Authorization" not in data

    @conftest.test_all_backends
    async def test_redirect_cross_host_no_remove_headers(self, backend, anyio_backend):
        with PoolManager(backend=backend) as http:
            r = await http.request(
                "GET",
                "%s/redirect" % self.base_url,
                fields={"target": "%s/headers" % self.base_url_alt},
                headers={"Authorization": "foo"},
                retries=Retry(remove_headers_on_redirect=[]),
            )

            assert r.status == 200

            data = json.loads(r.data.decode("utf-8"))

            assert data["Authorization"] == "foo"

    @conftest.test_all_backends
    async def test_redirect_cross_host_set_removed_headers(
        self, backend, anyio_backend
    ):
        with PoolManager(backend=backend) as http:
            r = await http.request(
                "GET",
                "%s/redirect" % self.base_url,
                fields={"target": "%s/headers" % self.base_url_alt},
                headers={"X-API-Secret": "foo", "Authorization": "bar"},
                retries=Retry(remove_headers_on_redirect=["X-API-Secret"]),
            )

            assert r.status == 200

            data = json.loads(r.data.decode("utf-8"))

            assert "X-API-Secret" not in data
            assert data["Authorization"] == "bar"

            r = await http.request(
                "GET",
                "%s/redirect" % self.base_url,
                fields={"target": "%s/headers" % self.base_url_alt},
                headers={"x-api-secret": "foo", "authorization": "bar"},
                retries=Retry(remove_headers_on_redirect=["X-API-Secret"]),
            )

            assert r.status == 200

            data = json.loads(r.data.decode("utf-8"))

            assert "x-api-secret" not in data
            assert "X-API-Secret" not in data
            assert data["Authorization"] == "bar"

    @conftest.test_all_backends
    async def test_raise_on_redirect(self, backend, anyio_backend):
        with PoolManager(backend=backend) as http:
            r = await http.request(
                "GET",
                "%s/redirect" % self.base_url,
                fields={
                    "target": "%s/redirect?target=%s/" % (self.base_url, self.base_url)
                },
                retries=Retry(total=None, redirect=1, raise_on_redirect=False),
            )

            assert r.status == 303

    @conftest.test_all_backends
    async def test_raise_on_status(self, backend, anyio_backend):
        with PoolManager(backend=backend) as http:
            with pytest.raises(MaxRetryError):
                # the default is to raise
                r = await http.request(
                    "GET",
                    "%s/status" % self.base_url,
                    fields={"status": "500 Internal Server Error"},
                    retries=Retry(total=1, status_forcelist=range(500, 600)),
                )

            with pytest.raises(MaxRetryError):
                # raise explicitly
                r = await http.request(
                    "GET",
                    "%s/status" % self.base_url,
                    fields={"status": "500 Internal Server Error"},
                    retries=Retry(
                        total=1, status_forcelist=range(500, 600), raise_on_status=True
                    ),
                )

            # don't raise
            r = await http.request(
                "GET",
                "%s/status" % self.base_url,
                fields={"status": "500 Internal Server Error"},
                retries=Retry(
                    total=1, status_forcelist=range(500, 600), raise_on_status=False
                ),
            )

            assert r.status == 500

    @pytest.mark.parametrize(
        ["target", "expected_target"],
        [
            ("/echo_uri?q=1#fragment", b"/echo_uri?q=1"),
            ("/echo_uri?#", b"/echo_uri?"),
            ("/echo_uri#?", b"/echo_uri"),
            ("/echo_uri#?#", b"/echo_uri"),
            ("/echo_uri??#", b"/echo_uri??"),
            ("/echo_uri?%3f#", b"/echo_uri?%3F"),
            ("/echo_uri?%3F#", b"/echo_uri?%3F"),
            ("/echo_uri?[]", b"/echo_uri?%5B%5D"),
        ],
    )
    @conftest.test_all_backends
    async def test_encode_http_target(
        self, target, expected_target, backend, anyio_backend
    ):
        with PoolManager() as http:
            url = "http://%s:%d%s" % (self.host, self.port, target)
            r = await http.request("GET", url)
            assert r.data == expected_target

    @conftest.test_all_backends
    async def test_missing_port(self, backend, anyio_backend):
        # Can a URL that lacks an explicit port like ':80' succeed, or
        # will all such URLs fail with an error?

        with PoolManager(backend=backend) as http:
            # By globally adjusting `DEFAULT_PORTS` we pretend for a moment
            # that HTTP's default port is not 80, but is the port at which
            # our test server happens to be listening.
            DEFAULT_PORTS["http"] = self.port
            try:
                r = await http.request("GET", "http://%s/" % self.host, retries=0)
            finally:
                DEFAULT_PORTS["http"] = 80

            assert r.status == 200
            assert r.data == b"Dummy server!"

    @conftest.test_all_backends
    async def test_headers(self, backend, anyio_backend):
        with PoolManager(backend=backend, headers={"Foo": "bar"}) as http:
            r = await http.request("GET", "%s/headers" % self.base_url)
            returned_headers = json.loads(r.data.decode())
            assert returned_headers.get("Foo") == "bar"

            r = await http.request("POST", "%s/headers" % self.base_url)
            returned_headers = json.loads(r.data.decode())
            assert returned_headers.get("Foo") == "bar"

            r = await http.request_encode_url("GET", "%s/headers" % self.base_url)
            returned_headers = json.loads(r.data.decode())
            assert returned_headers.get("Foo") == "bar"

            r = await http.request_encode_body("POST", "%s/headers" % self.base_url)
            returned_headers = json.loads(r.data.decode())
            assert returned_headers.get("Foo") == "bar"

            r = await http.request_encode_url(
                "GET", "%s/headers" % self.base_url, headers={"Baz": "quux"}
            )
            returned_headers = json.loads(r.data.decode())
            assert returned_headers.get("Foo") is None
            assert returned_headers.get("Baz") == "quux"

            r = await http.request_encode_body(
                "GET", "%s/headers" % self.base_url, headers={"Baz": "quux"}
            )
            returned_headers = json.loads(r.data.decode())
            assert returned_headers.get("Foo") is None
            assert returned_headers.get("Baz") == "quux"

    @conftest.test_all_backends
    async def test_http_with_ssl_keywords(self, backend, anyio_backend):
        with PoolManager(backend=backend, ca_certs="REQUIRED") as http:
            r = await http.request("GET", "http://%s:%s/" % (self.host, self.port))
            assert r.status == 200

    @conftest.test_all_backends
    async def test_http_with_ca_cert_dir(self, backend, anyio_backend):
        with PoolManager(
            backend=backend, ca_certs="REQUIRED", ca_cert_dir="/nosuchdir"
        ) as http:
            r = await http.request("GET", "http://%s:%s/" % (self.host, self.port))
            assert r.status == 200

    @conftest.test_all_backends
    async def test_cleanup_on_connection_error(self, backend, anyio_backend):
        """
        Test that connections are recycled to the pool on
        connection errors where no http response is received.
        """
        poolsize = 3

        with PoolManager(backend=backend, maxsize=poolsize, block=True) as http:
            pool = http.connection_from_host(self.host, self.port)
            assert pool.pool.qsize() == poolsize

            # force a connection error by supplying a non-existent
            # url. We won't get a response for this  and so the
            # conn won't be implicitly returned to the pool.
            url = "%s/redirect" % self.base_url
            with pytest.raises(MaxRetryError):
                await http.request("GET", url, fields={"target": "/"}, retries=0)

            r = await http.request("GET", url, fields={"target": "/"}, retries=1)
            r.release_conn()

            # the pool should still contain poolsize elements
            assert pool.pool.qsize() == poolsize


class TestFileUploads(HTTPDummyServerTestCase):
    @classmethod
    def setup_class(cls):
        super(TestFileUploads, cls).setup_class()
        cls.base_url = "http://%s:%d" % (cls.host, cls.port)
        cls.base_url_alt = "http://%s:%d" % (cls.host_alt, cls.port)

    @conftest.test_all_backends
    async def test_redirect_put_file(self, backend, anyio_backend):
        """PUT with file object should work with a redirection response"""
        retry = Retry(total=3, status_forcelist=[418])
        # httplib reads in 8k chunks; use a larger content length
        content_length = 65535
        data = b"A" * content_length
        uploaded_file = io.BytesIO(data)
        headers = {
            "test-name": "test_redirect_put_file",
            "Content-Length": str(content_length),
        }
        url = "%s/redirect?target=/echo&status=307" % self.base_url

        with PoolManager(backend=backend) as http:
            resp = await http.urlopen(
                "PUT", url, headers=headers, retries=retry, body=uploaded_file
            )
            assert resp.status == 200
            assert resp.data == data

    @conftest.test_all_backends
    async def test_retries_put_filehandle(self, backend, anyio_backend):
        """HTTP PUT retry with a file-like object should not timeout"""
        retry = Retry(total=3, status_forcelist=[418])
        # httplib reads in 8k chunks; use a larger content length
        content_length = 65535
        data = b"A" * content_length
        uploaded_file = io.BytesIO(data)
        headers = {
            "test-name": _test_name(
                "test_retries_put_filehandle", backend, anyio_backend
            ),
            "Content-Length": str(content_length),
        }

        with PoolManager(backend=backend) as http:
            resp = await http.urlopen(
                "PUT",
                "%s/successful_retry" % self.base_url,
                headers=headers,
                retries=retry,
                body=uploaded_file,
                redirect=False,
            )
            assert resp.status == 200

    @conftest.test_all_backends
    async def test_redirect_with_failed_tell(self, backend, anyio_backend):
        """Abort request if failed to get a position from tell()"""

        class BadTellObject(io.BytesIO):
            def tell(self):
                raise IOError

        body = BadTellObject(b"the data")
        url = "%s/redirect?target=/successful_retry" % self.base_url
        # httplib uses fileno if Content-Length isn't supplied,
        # which is unsupported by BytesIO.
        headers = {"Content-Length": "8"}

        with PoolManager(backend=backend) as http:
            with pytest.raises(UnrewindableBodyError) as e:
                await http.urlopen("PUT", url, headers=headers, body=body)
            assert "Unable to record file position for" in str(e.value)

    @conftest.test_all_backends
    async def test_redirect_with_failed_seek(self, backend, anyio_backend):
        """Abort request if failed to restore position with seek()"""

        class BadSeekObject(io.BytesIO):
            def seek(self, *_):
                raise IOError

        body = BadSeekObject(b"the data")
        url = "%s/redirect?target=/successful_retry" % self.base_url
        # httplib uses fileno if Content-Length isn't supplied,
        # which is unsupported by BytesIO.
        headers = {"Content-Length": "8"}

        with PoolManager(backend=backend) as http:
            with pytest.raises(UnrewindableBodyError) as e:
                await http.urlopen("PUT", url, headers=headers, body=body)
            assert (
                "An error occurred when rewinding request body for redirect/retry."
                == str(e.value)
            )


class TestRetry(HTTPDummyServerTestCase):
    @classmethod
    def setup_class(self):
        super(TestRetry, self).setup_class()
        self.base_url = "http://%s:%d" % (self.host, self.port)
        self.base_url_alt = "http://%s:%d" % (self.host_alt, self.port)

    @conftest.test_all_backends
    async def test_max_retry(self, backend, anyio_backend):
        with PoolManager(backend=backend) as http:
            with pytest.raises(MaxRetryError):
                await http.request(
                    "GET",
                    "%s/redirect" % self.base_url,
                    fields={"target": "/"},
                    retries=0,
                )

    @conftest.test_all_backends
    async def test_disabled_retry(self, backend, anyio_backend):
        """ Disabled retries should disable redirect handling. """
        with PoolManager(backend=backend) as http:
            r = await http.request(
                "GET",
                "%s/redirect" % self.base_url,
                fields={"target": "/"},
                retries=False,
            )
            assert r.status == 303

            r = await http.request(
                "GET",
                "%s/redirect" % self.base_url,
                fields={"target": "/"},
                retries=Retry(redirect=False),
            )
            assert r.status == 303

            with pytest.raises(NewConnectionError):
                await http.request(
                    "GET",
                    "http://thishostdoesnotexist.invalid/",
                    timeout=0.001,
                    retries=False,
                )

    @conftest.test_all_backends
    async def test_read_retries(self, backend, anyio_backend):
        """ Should retry for status codes in the whitelist """
        retry = Retry(read=1, status_forcelist=[418])

        with PoolManager(backend=backend) as http:
            resp = await http.request(
                "GET",
                "%s/successful_retry" % self.base_url,
                headers={
                    "test-name": _test_name("test_read_retries", backend, anyio_backend)
                },
                retries=retry,
            )
            assert resp.status == 200

    @conftest.test_all_backends
    async def test_read_total_retries(self, backend, anyio_backend):
        """ HTTP response w/ status code in the whitelist should be retried """
        headers = {
            "test-name": _test_name("test_read_total_retries", backend, anyio_backend)
        }
        retry = Retry(total=1, status_forcelist=[418])

        with PoolManager(backend=backend) as http:
            resp = await http.request(
                "GET",
                "%s/successful_retry" % self.base_url,
                headers=headers,
                retries=retry,
            )
            assert resp.status == 200

    @conftest.test_all_backends
    async def test_retries_wrong_whitelist(self, backend, anyio_backend):
        """HTTP response w/ status code not in whitelist shouldn't be retried"""
        headers = {
            "test-name": _test_name("test_wrong_whitelist", backend, anyio_backend)
        }
        retry = Retry(total=1, status_forcelist=[202])

        with PoolManager(backend=backend) as http:
            resp = await http.request(
                "GET",
                "%s/successful_retry" % self.base_url,
                retries=retry,
                headers=headers,
            )
            assert resp.status == 418

    @conftest.test_all_backends
    async def test_default_method_whitelist_retried(self, backend, anyio_backend):
        """Hip should retry methods in the default method whitelist"""
        headers = {
            "test-name": _test_name("test_default_whitelist", backend, anyio_backend)
        }
        retry = Retry(total=1, status_forcelist=[418])

        with PoolManager(backend=backend) as http:
            resp = await http.request(
                "OPTIONS",
                "%s/successful_retry" % self.base_url,
                headers=headers,
                retries=retry,
            )
            assert resp.status == 200

    @conftest.test_all_backends
    async def test_retries_wrong_method_list(self, backend, anyio_backend):
        """Method not in our whitelist should not be retried, even if code matches"""
        headers = {
            "test-name": _test_name(
                "test_wrong_method_whitelist", backend, anyio_backend
            )
        }
        retry = Retry(total=1, status_forcelist=[418], method_whitelist=["POST"])

        with PoolManager(backend=backend) as http:
            resp = await http.request(
                "GET",
                "%s/successful_retry" % self.base_url,
                headers=headers,
                retries=retry,
            )
            assert resp.status == 418

    @conftest.test_all_backends
    async def test_read_retries_unsuccessful(self, backend, anyio_backend):
        headers = {
            "test-name": _test_name(
                "test_read_retries_unsuccessful", backend, anyio_backend
            )
        }

        with PoolManager(backend=backend) as http:
            resp = await http.request(
                "GET", "%s/successful_retry" % self.base_url, headers=headers, retries=1
            )
            assert resp.status == 418

    @conftest.test_all_backends
    async def test_retry_reuse_safe(self, backend, anyio_backend):
        """ It should be possible to reuse a Retry object across requests """
        headers = {"test-name": _test_name("test_retry_safe", backend, anyio_backend)}
        retry = Retry(total=1, status_forcelist=[418])

        with PoolManager(backend=backend) as http:
            resp = await http.request(
                "GET",
                "%s/successful_retry" % self.base_url,
                headers=headers,
                retries=retry,
            )
            assert resp.status == 200
            resp = await http.request(
                "GET",
                "%s/successful_retry" % self.base_url,
                headers=headers,
                retries=retry,
            )
            assert resp.status == 200

    @conftest.test_all_backends
    async def test_retry_return_in_response(self, backend, anyio_backend):
        headers = {
            "test-name": _test_name(
                "test_retry_return_in_response", backend, anyio_backend
            )
        }
        retry = Retry(total=2, status_forcelist=[418])

        with PoolManager(backend=backend) as http:
            resp = await http.request(
                "GET",
                "%s/successful_retry" % self.base_url,
                headers=headers,
                retries=retry,
            )
            assert resp.status == 200
            assert resp.retries.total == 1
            assert resp.retries.history == (
                RequestHistory("GET", "/successful_retry", None, 418, None),
            )

    @conftest.test_all_backends
    async def test_retry_redirect_history(self, backend, anyio_backend):
        with PoolManager(backend=backend) as http:
            resp = await http.request(
                "GET", "%s/redirect" % self.base_url, fields={"target": "/"}
            )
            assert resp.status == 200
            assert resp.retries.history == (
                RequestHistory(
                    "GET", self.base_url + "/redirect?target=%2F", None, 303, "/"
                ),
            )

    @conftest.test_all_backends
    async def test_multi_redirect_history(self, backend, anyio_backend):
        with PoolManager(backend=backend) as http:
            r = await http.request(
                "GET",
                "%s/multi_redirect" % self.base_url,
                fields={"redirect_codes": "303,302,200"},
                redirect=False,
            )
            assert r.status == 303
            assert r.retries.history == tuple()

            r = await http.request(
                "GET",
                "%s/multi_redirect" % self.base_url,
                retries=10,
                fields={"redirect_codes": "303,302,301,307,302,200"},
            )
            assert r.status == 200
            assert r.data == b"Done redirecting"

            expected = [
                (303, "/multi_redirect?redirect_codes=302,301,307,302,200"),
                (302, "/multi_redirect?redirect_codes=301,307,302,200"),
                (301, "/multi_redirect?redirect_codes=307,302,200"),
                (307, "/multi_redirect?redirect_codes=302,200"),
                (302, "/multi_redirect?redirect_codes=200"),
            ]
            actual = [
                (history.status, history.redirect_location)
                for history in r.retries.history
            ]
            assert actual == expected


class TestRetryAfter(HTTPDummyServerTestCase):
    @classmethod
    def setup_class(self):
        super(TestRetryAfter, self).setup_class()
        self.base_url = "http://%s:%d" % (self.host, self.port)
        self.base_url_alt = "http://%s:%d" % (self.host_alt, self.port)

    @conftest.test_all_backends
    async def test_retry_after(self, backend, anyio_backend):
        url = "%s/retry_after" % self.base_url
        with PoolManager(backend=backend) as http:
            # Request twice in a second to get a 429 response.
            r = await http.request(
                "GET", url, fields={"status": "429 Too Many Requests"}, retries=False
            )
            r = await http.request(
                "GET", url, fields={"status": "429 Too Many Requests"}, retries=False
            )
            assert r.status == 429

            r = await http.request(
                "GET", url, fields={"status": "429 Too Many Requests"}, retries=True
            )
            assert r.status == 200

            # Request twice in a second to get a 503 response.
            r = await http.request(
                "GET", url, fields={"status": "503 Service Unavailable"}, retries=False
            )
            r = await http.request(
                "GET", url, fields={"status": "503 Service Unavailable"}, retries=False
            )
            assert r.status == 503

            r = await http.request(
                "GET", url, fields={"status": "503 Service Unavailable"}, retries=True
            )
            assert r.status == 200

            # Ignore Retry-After header on status which is not defined in
            # Retry.RETRY_AFTER_STATUS_CODES.
            r = await http.request(
                "GET", url, fields={"status": "418 I'm a teapot"}, retries=True
            )
            assert r.status == 418

    @conftest.test_all_backends
    async def test_redirect_after(self, backend, anyio_backend):
        with PoolManager(backend=backend) as http:
            r = await http.request(
                "GET", "%s/redirect_after" % self.base_url, retries=False
            )
            assert r.status == 303

            t = time.time()
            r = await http.request("GET", "%s/redirect_after" % self.base_url)
            assert r.status == 200
            delta = time.time() - t
            assert delta >= 1

            t = time.time()
            timestamp = t + 2
            r = await http.request(
                "GET", self.base_url + "/redirect_after?date=" + str(timestamp)
            )
            assert r.status == 200
            delta = time.time() - t
            assert delta >= 1

            # Retry-After is past
            t = time.time()
            timestamp = t - 1
            r = await http.request(
                "GET", self.base_url + "/redirect_after?date=" + str(timestamp)
            )
            delta = time.time() - t
            assert r.status == 200
            assert delta < 1


@pytest.mark.skipif(not HAS_IPV6, reason="IPv6 is not supported on this system")
class TestIPv6PoolManager(IPv6HTTPDummyServerTestCase):
    @classmethod
    def setup_class(cls):
        super(TestIPv6PoolManager, cls).setup_class()
        cls.base_url = "http://[%s]:%d" % (cls.host, cls.port)

    @conftest.test_all_backends
    async def test_ipv6(self, backend, anyio_backend):
        with PoolManager(backend=backend) as http:
            await http.request("GET", self.base_url)
# the default is to raise # raise explicitly # don't raise #fragment", b"/echo_uri?q=1"), #", b"/echo_uri?"), #?", b"/echo_uri"), #?#", b"/echo_uri"), #", b"/echo_uri??"), #", b"/echo_uri?%3F"), #", b"/echo_uri?%3F"), # Can a URL that lacks an explicit port like ':80' succeed, or # will all such URLs fail with an error? # By globally adjusting `DEFAULT_PORTS` we pretend for a moment # that HTTP's default port is not 80, but is the port at which # our test server happens to be listening. Test that connections are recycled to the pool on         connection errors where no http response is received. # force a connection error by supplying a non-existent # url. We won't get a response for this  and so the # conn won't be implicitly returned to the pool. # the pool should still contain poolsize elements PUT with file object should work with a redirection response # httplib reads in 8k chunks; use a larger content length HTTP PUT retry with a file-like object should not timeout # httplib reads in 8k chunks; use a larger content length Abort request if failed to get a position from tell() # httplib uses fileno if Content-Length isn't supplied, # which is unsupported by BytesIO. Abort request if failed to restore position with seek() # httplib uses fileno if Content-Length isn't supplied, # which is unsupported by BytesIO. Disabled retries should disable redirect handling. Should retry for status codes in the whitelist HTTP response w/ status code in the whitelist should be retried HTTP response w/ status code not in whitelist shouldn't be retried Hip should retry methods in the default method whitelist Method not in our whitelist should not be retried, even if code matches It should be possible to reuse a Retry object across requests # Request twice in a second to get a 429 response. # Request twice in a second to get a 503 response. # Ignore Retry-After header on status which is not defined in # Retry.RETRY_AFTER_STATUS_CODES. # Retry-After is past