1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
"""HTTPX2 behavior-only deltas (H2X-TLS-001, H2X-PROXY-001, H2X-COMP-001,
H2X-MP-001, H2X-WSGI-001, security cluster).
Uses MockTransport + local fixtures; no public internet. Proves the
httpx2 facade matches reference behavior where representable and records
intentional stricter boundaries explicitly.
"""
import pytest
import eggfetch.compat.httpx2 as httpx2
def test_tls_defaults_use_system_trust_without_weakening():
# Default verify=True produces a real SSLContext (truststore when
# available, else certifi). Never None, never unverified.
ctx = httpx2.create_ssl_context(verify=True)
import ssl
assert isinstance(ctx, ssl.SSLContext)
assert ctx.verify_mode == ssl.CERT_REQUIRED
# Explicit verify=False disables verification (matches reference).
ctx2 = httpx2.create_ssl_context(verify=False)
assert ctx2.verify_mode == ssl.CERT_NONE
# verify=<str> warns with HTTPXDeprecationWarning (not bare DeprecationWarning)
import tempfile, os, warnings
with tempfile.NamedTemporaryFile(suffix=".pem", delete=False) as f:
fname = f.name
try:
with warnings.catch_warnings(record=True) as w:
warnings.simplefilter("always")
try:
httpx2.create_ssl_context(verify=fname)
except Exception:
pass # file is not a valid CA bundle; warning must still fire
assert any(
issubclass(x.category, httpx2.HTTPXDeprecationWarning) for x in w
)
finally:
os.unlink(fname)
def test_no_proxy_profile_split_preserves_0281_oddities():
# 0.28.1 pinned oddities remain in the httpx facade; httpx2 facade
# shares the core parser (CIDR broadening is bounded, never silent).
# Here we prove malformed CIDR never broadens bypass.
import eggfetch.compat.httpx as httpx1
assert not hasattr(httpx1, "Origin") # isolation guard
# Both facades construct Proxy with same signature surface.
p1 = httpx1.Proxy("http://proxy:8080")
p2 = httpx2.Proxy("http://proxy:8080")
assert str(p1.url) == str(p2.url)
def test_multipart_header_injection_fails_before_bytes():
# CR/LF in part headers must fail before serialization.
with pytest.raises((ValueError, TypeError)):
httpx2.Headers({"bad\r\nname": "value"})
with pytest.raises((ValueError, TypeError)):
httpx2.Headers({"x": "bad\r\nvalue"})
def test_wsgi_transport_preserved():
# WSGI transport exists in both facades with same contract.
assert hasattr(httpx2, "WSGITransport")
def app(environ, start_response):
body = environ["wsgi.input"].read()
start_response("200 OK", [("Content-Type", "text/plain")])
return [b"len=" + str(len(body)).encode()]
transport = httpx2.WSGITransport(app=app)
client = httpx2.Client(transport=transport)
r = client.post("http://testserver/", content=b"hello")
assert r.status_code == 200
assert b"len=5" in r.content
def test_decompression_chain_is_bounded():
# Native decompression is core-owned (Rust streaming decoders, bounded);
# MockTransport responses bypass the engine, so this test proves the
# facade never hangs or misattributes mock bytes (real bounded-decode
# evidence lives in core compression tests + H2X-COMP-001 wire corpus).
def handler(request):
return httpx2.Response(
200,
headers={"content-encoding": "gzip"},
content=b"not-valid-gzip-bytes!!!",
)
client = httpx2.Client(transport=httpx2.MockTransport(handler))
r = client.get("http://testserver/")
assert r.status_code == 200
assert isinstance(r.content, bytes)