Merge branch 'master' of github.com:kevin1024/vcrpy into fix-resource-warning-2

tests/integration/test_httpx.py

@@ -1,7 +1,11 @@
+import os
+
 import pytest
 
 import vcr
 
+from ..assertions import assert_is_json_bytes
+
 asyncio = pytest.importorskip("asyncio")
 httpx = pytest.importorskip("httpx")
 
@@ -219,22 +223,6 @@ def test_redirect(httpbin, yml, do_request):
     assert cassette_response.request.headers.items() == response.request.headers.items()
 
 
-@pytest.mark.online
-def test_work_with_gzipped_data(httpbin, do_request, yml):
-    url = httpbin.url + "/gzip?foo=bar"
-    headers = {"accept-encoding": "deflate, gzip"}
-
-    with vcr.use_cassette(yml):
-        do_request(headers=headers)("GET", url)
-
-    with vcr.use_cassette(yml) as cassette:
-        cassette_response = do_request(headers=headers)("GET", url)
-
-        assert cassette_response.headers["content-encoding"] == "gzip"
-        assert cassette_response.read()
-        assert cassette.play_count == 1
-
-
 @pytest.mark.online
 @pytest.mark.parametrize("url", ["https://github.com/kevin1024/vcrpy/issues/" + str(i) for i in range(3, 6)])
 def test_simple_fetching(do_request, yml, url):
@@ -299,29 +287,75 @@ def test_stream(tmpdir, httpbin, do_request):
         assert cassette.play_count == 1
 
 
-@pytest.mark.online
-def test_text_content_type(tmpdir, httpbin, do_request):
-    url = httpbin.url + "/json"
+# Regular cassette formats support the status reason,
+# but the old HTTPX cassette format does not.
+@pytest.mark.parametrize(
+    "cassette_name,reason",
+    [
+        ("requests", "great"),
+        ("httpx_old_format", "OK"),
+    ],
+)
+def test_load_cassette_format(do_request, cassette_name, reason):
+    mydir = os.path.dirname(os.path.realpath(__file__))
+    yml = f"{mydir}/cassettes/gzip_{cassette_name}.yaml"
+    url = "https://httpbin.org/gzip"
 
-    with vcr.use_cassette(str(tmpdir.join("json_type.yaml"))):
-        response = do_request()("GET", url)
-
-    with vcr.use_cassette(str(tmpdir.join("json_type.yaml"))) as cassette:
+    with vcr.use_cassette(yml) as cassette:
         cassette_response = do_request()("GET", url)
-        assert cassette_response.content == response.content
+        assert str(cassette_response.request.url) == url
         assert cassette.play_count == 1
-        assert isinstance(cassette.responses[0]["content"], str)
+
+        # Should be able to load up the JSON inside,
+        # regardless whether the content is the gzipped
+        # in the cassette or not.
+        json = cassette_response.json()
+        assert json["method"] == "GET", json
+        assert cassette_response.status_code == 200
+        assert cassette_response.reason_phrase == reason
 
 
-@pytest.mark.online
-def test_binary_content_type(tmpdir, httpbin, do_request):
-    url = httpbin.url + "/bytes/1024"
+def test_gzip__decode_compressed_response_false(tmpdir, httpbin, do_request):
+    """
+    Ensure that httpx is able to automatically decompress the response body.
+    """
+    for _ in range(2):  # one for recording, one for re-playing
+        with vcr.use_cassette(str(tmpdir.join("gzip.yaml"))) as cassette:
+            response = do_request()("GET", httpbin + "/gzip")
+            assert response.headers["content-encoding"] == "gzip"  # i.e. not removed
+            # The content stored in the cassette should be gzipped.
+            assert cassette.responses[0]["body"]["string"][:2] == b"\x1f\x8b"
+            assert_is_json_bytes(response.content)  # i.e. uncompressed bytes
 
-    with vcr.use_cassette(str(tmpdir.join("json_type.yaml"))):
-        response = do_request()("GET", url)
 
-    with vcr.use_cassette(str(tmpdir.join("json_type.yaml"))) as cassette:
-        cassette_response = do_request()("GET", url)
-        assert cassette_response.content == response.content
-        assert cassette.play_count == 1
-        assert isinstance(cassette.responses[0]["content"], bytes)
+def test_gzip__decode_compressed_response_true(do_request, tmpdir, httpbin):
+    url = httpbin + "/gzip"
+
+    expected_response = do_request()("GET", url)
+    expected_content = expected_response.content
+    assert expected_response.headers["content-encoding"] == "gzip"  # self-test
+
+    with vcr.use_cassette(
+        str(tmpdir.join("decode_compressed.yaml")),
+        decode_compressed_response=True,
+    ) as cassette:
+        r = do_request()("GET", url)
+        assert r.headers["content-encoding"] == "gzip"  # i.e. not removed
+        content_length = r.headers["content-length"]
+        assert r.content == expected_content
+
+    # Has the cassette body been decompressed?
+    cassette_response_body = cassette.responses[0]["body"]["string"]
+    assert isinstance(cassette_response_body, str)
+
+    # Content should be JSON.
+    assert cassette_response_body[0:1] == "{"
+
+    with vcr.use_cassette(str(tmpdir.join("decode_compressed.yaml")), decode_compressed_response=True):
+        r = httpx.get(url)
+        assert "content-encoding" not in r.headers  # i.e. removed
+        assert r.content == expected_content
+
+        # As the content is uncompressed, it should have a bigger
+        # length than the compressed version.
+        assert r.headers["content-length"] > content_length