httpx transparently inflates gzip/deflate response bodies (it keeps the
Content-Encoding header but .content/.text are the decoded plaintext). Three
more tests asserted on the body of a compressed response and so were wrong:
- pr43939 / pr17629: re-POSTed r.content (decoded plaintext, not the gzip
stream) through the inflate input filter -- same round-trip bug as
pr49328/deflate.
- reflector: 'assert r.text != payload' could never hold, since httpx
decoded the gzip body straight back to the payload.
Generalise the GET_RAW helper into a raw_response(method, path, ...) that
streams the response, reads iter_raw() (undecoded bytes), and stashes them on
.raw_content while leaving .status_code/.headers intact -- so tests can assert
on both the raw body and the headers. GET_RAW now delegates to it; update the
three tests to use raw_response/.raw_content.
The length-based tests (rwrite, passbrigade, getfile, byterange7) are
unaffected: they measure len(.content) after decoding, which equals the
original length regardless of transfer compression.
git-svn-id: https://svn.apache.org/repos/asf/httpd/httpd/trunk@
1934950 13f79535-47bb-0310-9956-
ffa450edef68
def GET(self, path: str, **kwargs: object) -> httpx.Response:
return self._request("GET", path, **kwargs)
- def _raw_body(
+ def raw_response(
self,
method: str,
path: str,
*,
cert: str | None = None,
**kwargs: object,
- ) -> bytes:
- """Return the response body WITHOUT content-decoding.
+ ) -> httpx.Response:
+ """Like :meth:`_request` but WITHOUT content-decoding the body.
httpx transparently inflates gzip/deflate responses, so ``.content`` is
- the decoded plaintext. The mod_deflate round-trip tests need the raw
- compressed bytes (to re-POST them through the inflate input filter), so
- stream the response and read ``iter_raw()``, which yields the bytes as
- they came off the wire -- the analog of LWP not auto-decoding.
+ the decoded plaintext (while ``Content-Encoding`` is left in place).
+ Tests that need the bytes exactly as they came off the wire -- e.g. the
+ mod_deflate round-trips that re-POST the gzip through an inflate filter,
+ or mod_reflector asserting the body was actually transformed -- can't use
+ that. Stream the response, read ``iter_raw()`` (the undecoded bytes, the
+ analog of LWP not auto-decoding), and stash them on ``.raw_content`` so
+ callers still see ``.status_code`` and ``.headers``.
"""
client = self._client_for(cert) if cert is not None else self._client
request = client.build_request(method, self._url(path), **kwargs) # type: ignore[arg-type]
response = client.send(request, stream=True)
try:
- return b"".join(response.iter_raw())
+ response.raw_content = b"".join(response.iter_raw()) # type: ignore[attr-defined]
finally:
response.close()
+ return response
def GET_RAW(self, path: str, **kwargs: object) -> bytes:
"""GET ``path`` and return the raw, undecoded response body bytes."""
- return self._raw_body("GET", path, **kwargs)
+ return self.raw_response("GET", path, **kwargs).raw_content # type: ignore[attr-defined]
def HEAD(self, path: str, **kwargs: object) -> httpx.Response:
return self._request("HEAD", path, **kwargs)
content = http.GET_BODY(URI)
assert t_cmp(content, EXPECTED)
- r = http.GET(URI, headers={"Accept-Encoding": "gzip"})
+ # raw_response keeps the gzip stream undecoded (httpx would auto-decompress
+ # .content) so we can re-POST it through the inflate input filter.
+ r = http.raw_response("GET", URI, headers={"Accept-Encoding": "gzip"})
assert t_cmp(r.status_code, 200)
renc = r.headers.get("Content-Encoding", "")
assert t_cmp(renc, "gzip"), "response was gzipped"
deflated = http.POST_BODY(
- INFLATOR, content=r.content, headers={"Content-Encoding": "gzip"}
+ INFLATOR, content=r.raw_content, headers={"Content-Encoding": "gzip"}
)
assert t_cmp(deflated, EXPECTED)
content = http.GET_BODY(URI)
assert t_cmp(content, EXPECTED)
- r = http.GET(URI, headers={"Accept-Encoding": "gzip"})
+ # raw_response keeps the gzip stream undecoded (httpx would auto-decompress
+ # .content) so we can re-POST it through the inflate input filter.
+ r = http.raw_response("GET", URI, headers={"Accept-Encoding": "gzip"})
assert t_cmp(r.status_code, 200)
renc = r.headers.get("Content-Encoding", "")
assert t_cmp(renc, "gzip"), "response was gzipped"
deflated = http.POST_BODY(
- INFLATOR, content=r.content, headers={"Content-Encoding": "gzip"}
+ INFLATOR, content=r.raw_content, headers={"Content-Encoding": "gzip"}
)
assert t_cmp(deflated, EXPECTED)
@need_module("mod_reflector", "mod_deflate")
@pytest.mark.parametrize("url,payload", TESTCASES, ids=lambda v: str(v))
def test_reflector(http, url, payload):
- r = http.POST(url, content=payload, headers=HEADERS)
+ # raw_response keeps the body undecoded: the deflate case must observe that
+ # the gzip body differs from the payload, but httpx would auto-decompress
+ # .text/.content back to the payload and mask the transformation.
+ r = http.raw_response("POST", url, content=payload, headers=HEADERS)
+ body = r.raw_content.decode("latin-1")
assert t_cmp(r.status_code, 200), "Checking return code is '200'"
if "_nodeflate" in url:
# With no filter, we should receive what we have sent.
- assert t_cmp(r.text, payload)
+ assert t_cmp(body, payload)
assert t_cmp(r.headers.get("Content-Encoding"), None), \
"'Content-Encoding' has not been added because there was no filter"
else:
# With DEFLATE, input was updated and 'Content-Encoding' added.
- assert r.text != payload
+ assert body != payload
assert t_cmp(r.headers.get("Content-Encoding"), "gzip"), \
"'Content-Encoding' has been added by the DEFLATE filter"
projects / thirdparty / apache / httpd.git / commitdiff
