import pytest
from eggfetch.compat.httpx import (
AsyncClient,
Client,
Cookies,
MockTransport,
Request,
Response,
Headers,
)
class TestCookieScope:
def test_domain_path_scoped_get(self):
c = Cookies()
c.set("session", "val1", domain=".example.com", path="/")
c.set("session", "val2", domain=".api.example.com", path="/")
assert c.get("session", domain=".example.com") == "val1"
assert c.get("session", domain=".api.example.com") == "val2"
def test_cookie_conflict(self):
from eggfetch.compat.httpx._exceptions import CookieConflict
c = Cookies()
c.set("session", "val1", domain=".example.com", path="/")
c.set("session", "val2", domain=".example.com", path="/api")
with pytest.raises(CookieConflict):
c.get("session")
def test_cookie_conflict_resolved_with_scope(self):
c = Cookies()
c.set("session", "val1", domain=".example.com", path="/")
c.set("session", "val2", domain=".example.com", path="/api")
assert c.get("session", domain=".example.com", path="/") == "val1"
assert c.get("session", domain=".example.com", path="/api") == "val2"
def test_delete_by_domain_path(self):
c = Cookies()
c.set("session", "val1", domain=".example.com", path="/")
c.set("session", "val2", domain=".example.com", path="/api")
c.delete("session", domain=".example.com", path="/")
assert c.get("session", domain=".example.com", path="/") is None
assert c.get("session", domain=".example.com", path="/api") == "val2"
def test_clear_domain_path(self):
c = Cookies()
c.set("a", "1", domain=".example.com", path="/")
c.set("b", "2", domain=".example.com", path="/api")
c.clear(domain=".example.com", path="/")
assert c.get("a", domain=".example.com", path="/") is None
assert c.get("b", domain=".example.com", path="/api") == "2"
class TestMultipleSetCookieHeaders:
def test_multiple_set_cookie(self):
h = Headers([
("Set-Cookie", "a=1; Path=/"),
("Set-Cookie", "b=2; Path=/api"),
])
resp = Response(200, headers=h, request=Request("GET", "https://example.com"))
c = Cookies()
c.extract_cookies(resp)
assert c.get("a") == "1"
assert c.get("b") == "2"
def test_set_cookie_with_attributes(self):
h = Headers([
("Set-Cookie", "session=abc; Domain=.example.com; Path=/; Secure; HttpOnly"),
])
resp = Response(200, headers=h, request=Request("GET", "https://example.com"))
c = Cookies()
c.extract_cookies(resp)
assert c.get("session") == "abc"
class TestCookieInSendLoop:
def test_cookies_set_by_response_available_next_hop(self):
def handler(request):
if request.url.path == "/set":
return Response(
200,
headers=[("Set-Cookie", "token=abc123; Path=/")],
text="ok",
)
cookie = request.headers.get("cookie", "none")
return Response(200, text=cookie)
with Client(transport=MockTransport(handler)) as c:
resp1 = c.get("http://testserver/set")
assert resp1.text == "ok"
assert c.cookies.get("token") == "abc123"
resp2 = c.get("http://testserver/check")
assert "token=abc123" in resp2.text
def test_redirect_sets_cookie_available_after(self):
def handler(request):
if request.url.path == "/redirect":
return Response(
302,
headers=[
("Location", "/target"),
("Set-Cookie", "redirect_cookie=yes; Path=/"),
],
)
cookie = request.headers.get("cookie", "none")
return Response(200, text=cookie)
with Client(transport=MockTransport(handler), follow_redirects=True) as c:
resp = c.get("http://testserver/redirect")
assert "redirect_cookie=yes" in resp.text
def test_response_cookies_not_client_jar(self):
def handler(request):
return Response(
200,
headers=[("Set-Cookie", "resp_only=xyz; Path=/")],
text="ok",
)
with Client(transport=MockTransport(handler)) as c:
c.cookies.set("pre_existing", "val")
resp = c.get("http://testserver/")
assert resp.cookies.get("resp_only") == "xyz"
def test_client_cookies_mutation_affects_next_request(self):
def handler(request):
cookie = request.headers.get("cookie", "none")
return Response(200, text=cookie)
with Client(transport=MockTransport(handler)) as c:
resp1 = c.get("http://testserver/")
assert resp1.text == "none"
c.cookies.set("added", "later")
resp2 = c.get("http://testserver/")
assert "added=later" in resp2.text
def test_auth_redirect_cookie_propagation(self):
from eggfetch.compat.httpx import Auth
class CookieAuth(Auth):
def auth_flow(self, request):
request.headers["authorization"] = "Bearer token"
response = yield request
def handler(request):
if request.url.path == "/target":
return Response(200, text="ok")
if request.headers.get("authorization") == "Bearer token":
return Response(
302,
headers=[
("Location", "/target"),
("Set-Cookie", "auth_cookie=granted; Path=/"),
],
)
return Response(401)
with Client(
transport=MockTransport(handler),
auth=CookieAuth(),
follow_redirects=True,
) as c:
resp = c.get("http://testserver/protected")
assert c.cookies.get("auth_cookie") == "granted"
class TestCookieJarConstruction:
def test_from_dict(self):
c = Cookies({"a": "1", "b": "2"})
assert c.get("a") == "1"
assert c.get("b") == "2"
def test_from_list(self):
c = Cookies([("a", "1"), ("b", "2")])
assert c.get("a") == "1"
assert c.get("b") == "2"
def test_from_cookies(self):
original = Cookies({"a": "1"})
copy = Cookies(original)
assert copy.get("a") == "1"
def test_copy_independence(self):
original = Cookies({"a": "1"})
copy = Cookies(original)
copy.set("b", "2")
assert original.get("b") is None
def test_invalid_type(self):
with pytest.raises(TypeError):
Cookies(123)
class TestAsyncCookieScope:
@pytest.mark.asyncio
async def test_cookies_set_by_response_available_next_hop_async(self):
async def handler(request):
if request.url.path == "/set":
return Response(
200,
headers=[("Set-Cookie", "token=abc123; Path=/")],
text="ok",
)
cookie = request.headers.get("cookie", "none")
return Response(200, text=cookie)
async with AsyncClient(async_transport=MockTransport(handler)) as c:
resp1 = await c.get("http://testserver/set")
assert resp1.text == "ok"
assert c.cookies.get("token") == "abc123"
resp2 = await c.get("http://testserver/check")
assert "token=abc123" in resp2.text
@pytest.mark.asyncio
async def test_redirect_cookie_propagation_async(self):
async def handler(request):
if request.url.path == "/redirect":
return Response(
302,
headers=[
("Location", "/target"),
("Set-Cookie", "redirect_cookie=yes; Path=/"),
],
)
cookie = request.headers.get("cookie", "none")
return Response(200, text=cookie)
async with AsyncClient(
async_transport=MockTransport(handler), follow_redirects=True
) as c:
resp = await c.get("http://testserver/redirect")
assert "redirect_cookie=yes" in resp.text