Make session.cookies the only cookie store and follow redirects in Python

tests/integration/test_httpbin.py

@@ -1,12 +1,13 @@
+import asyncio
 import base64
 import socket
 from collections.abc import Iterator
 
 import pytest
 from httpbin.core import resource as httpbin_resource
 
-from tls_client import Session
-from tls_client.exceptions import TLSClientException
+from tls_client import AsyncSession, Session
+from tls_client.exceptions import TLSClientException, TooManyRedirects
 
 pytestmark = pytest.mark.integration
 
@@ -118,11 +119,61 @@ def test_not_followed_by_default(self, httpbin, session):
         assert response.status_code == 302
         assert response.headers["Location"] == "/get"
 
-    def test_followed_when_allowed(self, httpbin, session):
-        response = session.get(f"{httpbin.url}/redirect/2", allow_redirects=True)
+    @pytest.mark.parametrize("path", ["/redirect/3", "/relative-redirect/3", "/absolute-redirect/3"])
+    def test_followed_when_allowed(self, httpbin, session, path):
+        response = session.get(f"{httpbin.url}{path}", allow_redirects=True)
 
         assert response.status_code == 200
         assert response.url == f"{httpbin.url}/get"
+        assert [r.status_code for r in response.history] == [302, 302, 302]
+        assert response.history[0].url == f"{httpbin.url}{path}"
+
+    def test_redirect_to_a_relative_url(self, httpbin, session):
+        response = session.get(f"{httpbin.url}/redirect-to", params={"url": "/get?a=1"}, allow_redirects=True)
+
+        assert response.url == f"{httpbin.url}/get?a=1"
+        assert response.json()["args"] == {"a": "1"}
+
+    @pytest.mark.parametrize(
+        ("status", "method"), [(301, "GET"), (302, "GET"), (303, "GET"), (307, "POST"), (308, "POST")]
+    )
+    def test_post_is_resent_as_the_status_requires(self, httpbin, session, status, method):
+        response = session.post(
+            f"{httpbin.url}/redirect-to",
+            params={"url": "/anything", "status_code": status},
+            data={"a": "1"},
+            allow_redirects=True,
+        )
+
+        echoed = response.json()
+        assert echoed["method"] == method
+        if method == "POST":
+            assert echoed["form"] == {"a": "1"}
+            assert echoed["headers"]["Content-Type"] == "application/x-www-form-urlencoded"
+        else:
+            assert echoed["form"] == {}
+            assert "Content-Type" not in echoed["headers"]
+
+    def test_too_many_redirects_raise(self, httpbin, session):
+        with pytest.raises(TooManyRedirects):
+            session.get(f"{httpbin.url}/redirect/31", allow_redirects=True)
+
+    def test_credentials_and_cookies_stay_with_their_host(self, httpbin, session):
+        session.get(f"{httpbin.url}/cookies/set", params={"flavor": "oat"})
+        other_host = httpbin.url.replace("127.0.0.1", "localhost")
+
+        response = session.get(
+            f"{httpbin.url}/redirect-to",
+            params={"url": f"{other_host}/headers"},
+            headers={"Authorization": "Bearer secret", "X-Custom": "1"},
+            allow_redirects=True,
+        )
+
+        received = response.json()["headers"]
+        assert received["Host"] == other_host.removeprefix("http://")
+        assert received["X-Custom"] == "1"
+        assert "Authorization" not in received
+        assert "Cookie" not in received
 
 
 class TestCookies:
@@ -141,6 +192,46 @@ def test_only_cookies_matching_the_host_are_sent(self, httpbin, session):
 
         assert response.json() == {"cookies": {"host_only": "1", "per_request": "3"}}
 
+    def test_cookie_set_by_a_redirect_is_stored_and_sent_to_its_target(self, httpbin, session):
+        # /cookies/set answers with the cookie and a redirect to /cookies, which echoes the cookies it receives.
+        response = session.get(f"{httpbin.url}/cookies/set", params={"flavor": "oat"}, allow_redirects=True)
+
+        assert response.history[0].status_code == 302
+        assert response.json() == {"cookies": {"flavor": "oat"}}
+        assert response.cookies["flavor"] == "oat"
+        assert session.cookies["flavor"] == "oat"
+
+    def test_cookie_set_by_a_redirect_reaches_the_async_session_jar(self, httpbin):
+        async def main():
+            async with AsyncSession() as session:
+                await session.get(f"{httpbin.url}/cookies/set", params={"flavor": "oat"}, allow_redirects=True)
+                return session.cookies.get_dict()
+
+        assert asyncio.run(main()) == {"flavor": "oat"}
+
+    def test_cleared_cookies_are_no_longer_sent(self, httpbin, session):
+        session.get(f"{httpbin.url}/cookies/set", params={"flavor": "oat"})
+        assert session.get(f"{httpbin.url}/cookies").json() == {"cookies": {"flavor": "oat"}}
+
+        session.cookies.clear()
+
+        assert session.get(f"{httpbin.url}/cookies").json() == {"cookies": {}}
+
+    def test_removed_cookie_is_no_longer_sent(self, httpbin, session):
+        session.get(f"{httpbin.url}/cookies/set", params={"flavor": "oat", "size": "big"})
+
+        del session.cookies["flavor"]
+
+        assert session.get(f"{httpbin.url}/cookies").json() == {"cookies": {"size": "big"}}
+
+    def test_cookie_deleted_by_the_server_is_no_longer_sent(self, httpbin, session):
+        session.get(f"{httpbin.url}/cookies/set", params={"flavor": "oat", "size": "big"})
+
+        session.get(f"{httpbin.url}/cookies/delete", params={"flavor": ""})
+
+        assert "flavor" not in session.cookies
+        assert session.get(f"{httpbin.url}/cookies").json() == {"cookies": {"size": "big"}}
+
 
 class TestFailures:
     def test_refused_connection_raises(self, session):

tests/integration/test_local_server.py

@@ -1,6 +1,7 @@
-"""Real requests through the native library against a stdlib HTTP server, for bodies and headers httpbin can't
-produce byte for byte."""
+"""Real requests through the native library against a stdlib HTTP server, for bodies, headers and redirects httpbin
+can't produce byte for byte."""
 
+import asyncio
 import base64
 import json
 import os
@@ -11,7 +12,7 @@
 
 import pytest
 
-from tls_client import Session
+from tls_client import AsyncSession, Session
 from tls_client.exceptions import TLSClientException
 
 pytestmark = pytest.mark.integration
@@ -45,6 +46,12 @@ def do_GET(self) -> None:
         elif self.path == "/slow":
             time.sleep(SLOW_SECONDS)
             self.reply("text/plain", b"done")
+        elif self.path.startswith("/chain/"):
+            # /chain/<n> sets the cookie hop<n> and redirects to /chain/<n - 1>, /chain/0 to /headers.
+            hops = int(self.path.removeprefix("/chain/"))
+            self.redirect(f"/chain/{hops - 1}" if hops > 1 else "/headers", f"hop{hops}=set; Path=/")
+        elif self.path == "/expire-hop1":
+            self.redirect("/headers", "hop1=; Max-Age=0; Path=/")
         else:
             content_type, body = BODIES[self.path]
             self.reply(content_type, body, {"x-MiXeD-case": "yes"})
@@ -57,6 +64,13 @@ def echo_headers(self) -> None:
         received = {name.lower(): value for name, value in self.headers.items()}
         self.reply("application/json", json.dumps(received).encode())
 
+    def redirect(self, location: str, set_cookie: str) -> None:
+        self.send_response(302)
+        self.send_header("Location", location)
+        self.send_header("Set-Cookie", set_cookie)
+        self.send_header("Content-Length", "0")
+        self.end_headers()
+
     def reply(self, content_type: str, body: bytes, extra_headers: dict[str, str] | None = None) -> None:
         self.send_response(200)
         self.send_header("Content-Type", content_type)
@@ -163,3 +177,34 @@ def test_fractional_session_timeout_is_applied(self, server_url, session):
 
         with pytest.raises(TLSClientException, match="Timeout exceeded"):
             session.get(f"{server_url}/slow")
+
+
+class TestRedirectCookies:
+    def test_cookie_set_on_every_hop_is_stored_and_sent_on_the_next_one(self, server_url, session):
+        response = session.get(f"{server_url}/chain/3", allow_redirects=True)
+
+        assert [r.status_code for r in response.history] == [302, 302, 302]
+        assert response.url == f"{server_url}/headers"
+        # Cookies of the same path have no defined order in the header; it differs between Python versions.
+        assert sorted(response.json()["cookie"].split("; ")) == ["hop1=set", "hop2=set", "hop3=set"]
+        assert session.cookies.get_dict() == {"hop3": "set", "hop2": "set", "hop1": "set"}
+        assert response.cookies.get_dict() == {"hop3": "set", "hop2": "set", "hop1": "set"}
+
+    def test_cookie_expired_on_a_hop_is_not_sent_on_the_next_one(self, server_url, session):
+        session.get(f"{server_url}/chain/2", allow_redirects=True)
+
+        response = session.get(f"{server_url}/expire-hop1", allow_redirects=True)
+
+        assert response.json()["cookie"] == "hop2=set"
+        assert session.cookies.get_dict() == {"hop2": "set"}
+
+    def test_cookies_set_on_every_hop_end_up_in_the_async_session_jar(self, server_url):
+        async def main():
+            async with AsyncSession() as session:
+                await session.get(f"{server_url}/chain/2", allow_redirects=True)
+                return session.cookies.get_dict(), (await session.get(f"{server_url}/headers")).json()
+
+        jar, received = asyncio.run(main())
+
+        assert jar == {"hop2": "set", "hop1": "set"}
+        assert sorted(received["cookie"].split("; ")) == ["hop1=set", "hop2=set"]

tests/unit/test_async_sessions.py

@@ -13,7 +13,9 @@
 
 
 def sent_cookies(payload: dict) -> dict[str, str]:
-    return {cookie["name"]: cookie["value"] for cookie in payload["requestCookies"]}
+    """Cookies of the Cookie header the request was sent with."""
+    header = payload["headers"].get("Cookie", "")
+    return dict(pair.split("=", 1) for pair in header.split("; ") if pair)
 
 
 class CountingExecutor(ThreadPoolExecutor):
@@ -100,14 +102,15 @@ async def main():
         assert fake_lib.sent["requestBody"] == "raw"
 
     def test_keyword_arguments_are_forwarded(self, fake_lib):
+        fake_lib.reply_with(status=302, headers={"Location": ["/next"]})
+
         async def main():
             async with AsyncSession() as session:
                 await session.execute_request("GET", URL, params={"q": "1"}, allow_redirects=True)
 
         asyncio.run(main())
 
-        assert fake_lib.sent["requestUrl"] == f"{URL}?q=1"
-        assert fake_lib.sent["followRedirects"] is True
+        assert [payload["requestUrl"] for payload in fake_lib.payloads] == [f"{URL}?q=1", "https://example.com/next"]
 
     def test_requests_use_the_native_session_of_the_wrapped_session(self, fake_lib):
         async def main():
@@ -161,6 +164,37 @@ async def main():
         assert dict(session.cookies) == {"a": "1", "b": "2"}
         assert sent_cookies(fake_lib.sent) == {"a": "1", "b": "2"}
 
+    def test_cookies_set_on_every_redirect_hop_end_up_in_the_jar(self, fake_lib):
+        fake_lib.reply_with(status=302, headers={"Location": ["/second"], "Set-Cookie": ["a=1; Path=/"]})
+        fake_lib.reply_with(status=307, headers={"Location": ["/third"], "Set-Cookie": ["b=2; Path=/"]})
+        fake_lib.reply_with(headers={"Set-Cookie": ["c=3; Path=/"]})
+
+        async def main():
+            async with AsyncSession() as session:
+                response = await session.get(URL, allow_redirects=True)
+                await session.get(URL)
+                return session, response
+
+        session, response = asyncio.run(main())
+
+        assert len(response.history) == 2
+        assert dict(session.cookies) == {"a": "1", "b": "2", "c": "3"}
+        assert sent_cookies(fake_lib.payloads[2]) == {"a": "1", "b": "2"}
+        assert sent_cookies(fake_lib.sent) == {"a": "1", "b": "2", "c": "3"}
+
+    def test_cleared_cookies_are_not_sent(self, fake_lib):
+        fake_lib.reply_with(headers={"Set-Cookie": ["a=1; Path=/"]})
+
+        async def main():
+            async with AsyncSession() as session:
+                await session.get(URL)
+                session.cookies.clear()
+                await session.get(URL)
+
+        asyncio.run(main())
+
+        assert sent_cookies(fake_lib.sent) == {}
+
     @pytest.mark.parametrize("cookies", [{"from_request": "2"}, cookiejar_from_dict({"from_request": "2"})])
     def test_request_cookies_are_kept_in_the_jar_like_session_does(self, fake_lib, cookies):
         async def main():

tests/unit/test_cookies.py

@@ -196,7 +196,6 @@ def test_cookie_for_a_foreign_domain_is_rejected(self):
         assert len(response_jar) == 0
         assert len(session_jar) == 0
 
-    @pytest.mark.xfail(reason="expired cookies are dropped from the response jar but stay in the session jar")
     def test_server_can_delete_a_session_cookie(self):
         session_jar = RequestsCookieJar()
         extract_cookies_to_jar("https://example.com/", {}, session_jar, {"Set-Cookie": ["sid=abc; Path=/"]})