Use uv on CI

The asyncio function is deprecated starting from Python 3.14 and
will be removed from Python 3.16.

.github/workflows/codespell.yml

@@ -13,7 +13,7 @@ permissions:
 jobs:
   codespell:
     name: Check for spelling errors
-    runs-on: ubuntu-latest
+    runs-on: ubuntu-24.04
 
     steps:
       - name: Checkout

.github/workflows/docs.yml

@@ -7,7 +7,7 @@ on:
 
 jobs:
   validate:
-    runs-on: ubuntu-latest
+    runs-on: ubuntu-24.04
 
     steps:
       - uses: actions/checkout@v4

.github/workflows/main.yml

@@ -11,7 +11,7 @@ on:
 
 jobs:
   build:
-    runs-on: ubuntu-latest
+    runs-on: ubuntu-24.04
     strategy:
       fail-fast: false
       matrix:
@@ -37,6 +37,8 @@ jobs:
 
     steps:
       - uses: actions/checkout@v4
+      - name: Install uv
+        uses: astral-sh/setup-uv@v5
 
       - name: Set up Python ${{ matrix.python-version }}
         uses: actions/setup-python@v5
@@ -47,9 +49,16 @@ jobs:
 
       - name: Install project dependencies
         run: |
-          pip install --upgrade pip setuptools
-          pip install codecov '.[tests]' '${{ matrix.urllib3-requirement }}'
-          pip check
+          uv pip install --system --upgrade pip setuptools
+          uv pip install --system codecov '.[tests]' '${{ matrix.urllib3-requirement }}'
+          uv pip check
+
+      - name: Allow creation of user namespaces (e.g. to the unshare command)
+        run: |
+          # .. so that we don't get error:
+          # unshare: write failed /proc/self/uid_map: Operation not permitted
+          # Idea from https://github.com/YoYoGames/GameMaker-Bugs/issues/6015#issuecomment-2135552784 .
+          sudo sysctl kernel.apparmor_restrict_unprivileged_userns=0
 
       - name: Run online tests
         run: ./runtests.sh --cov=./vcr --cov-branch --cov-report=xml --cov-append -m online

.github/workflows/pre-commit-detect-outdated.yml

@@ -16,7 +16,7 @@ permissions:
 jobs:
   pre_commit_detect_outdated:
     name: Detect outdated pre-commit hooks
-    runs-on: ubuntu-latest
+    runs-on: ubuntu-24.04
     steps:
       - uses: actions/checkout@v4
 

.github/workflows/pre-commit.yml

@@ -11,7 +11,7 @@ on:
 jobs:
   pre-commit:
     name: Run pre-commit
-    runs-on: ubuntu-latest
+    runs-on: ubuntu-24.04
     steps:
     - uses: actions/checkout@v4
     - uses: actions/setup-python@v5

.pre-commit-config.yaml

@@ -3,7 +3,7 @@
 
 repos:
   - repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: v0.8.4
+    rev: v0.11.2
     hooks:
       - id: ruff
         args: ["--output-format=full"]

.readthedocs.yaml

@@ -7,7 +7,7 @@ version: 2
 
 # Set the version of Python and other tools you might need
 build:
-  os: ubuntu-22.04
+  os: ubuntu-24.04
   tools:
     python: "3.12"
 

docs/advanced.rst

@@ -427,3 +427,16 @@ If you want to save the cassette only when the test succeeds, set the Cassette
 
     # Since there was an exception, the cassette file hasn't been created.
     assert not os.path.exists('fixtures/vcr_cassettes/synopsis.yaml')
+
+Drop unused requests
+--------------------
+
+Even if any HTTP request is changed or removed from tests, previously recorded
+interactions remain in the cassette file.  If set the ``drop_unused_requests``
+option to ``True``, VCR will not save old HTTP interactions if they are not used.
+
+.. code:: python
+
+    my_vcr = VCR(drop_unused_requests=True)
+    with my_vcr.use_cassette('fixtures/vcr_cassettes/synopsis.yaml'):
+        ...  # your HTTP interactions here

docs/conf.py

@@ -316,5 +316,5 @@ texinfo_documents = [
 
 
 # Example configuration for intersphinx: refer to the Python standard library.
-intersphinx_mapping = {"https://docs.python.org/": None}
+intersphinx_mapping = {"python": ("https://docs.python.org/3", None)}
 html_theme = "alabaster"

docs/requirements.txt

@@ -1,2 +1,2 @@
 sphinx<9
-sphinx_rtd_theme==2.0.0
+sphinx_rtd_theme==3.0.2

pyproject.toml

@@ -1,19 +1,11 @@
 [tool.codespell]
 skip = '.git,*.pdf,*.svg,.tox'
 ignore-regex = "\\\\[fnrstv]"
-#
-# ignore-words-list = ''
 
 [tool.pytest.ini_options]
-addopts = [
-    "--strict-config",
-    "--strict-markers",
-]
+addopts = ["--strict-config", "--strict-markers"]
+asyncio_default_fixture_loop_scope = "function"
 markers = ["online"]
-filterwarnings = [
-    "error",
-    '''ignore:datetime\.datetime\.utcfromtimestamp\(\) is deprecated and scheduled for removal in a future version.*:DeprecationWarning''',
-]
 
 [tool.ruff]
 line-length = 110

tests/integration/test_config.py

@@ -5,6 +5,7 @@ from urllib.request import urlopen
 import pytest
 
 import vcr
+from vcr.cassette import Cassette
 
 
 @pytest.mark.online
@@ -85,3 +86,21 @@ def test_dont_record_on_exception(tmpdir, httpbin):
             assert b"Not in content" in urlopen(httpbin.url).read()
 
     assert not os.path.exists(str(tmpdir.join("dontsave2.yml")))
+
+
+def test_set_drop_unused_requests(tmpdir, httpbin):
+    my_vcr = vcr.VCR(drop_unused_requests=True)
+    file = str(tmpdir.join("test.yaml"))
+
+    with my_vcr.use_cassette(file):
+        urlopen(httpbin.url)
+        urlopen(httpbin.url + "/get")
+
+    cassette = Cassette.load(path=file)
+    assert len(cassette) == 2
+
+    with my_vcr.use_cassette(file):
+        urlopen(httpbin.url)
+
+    cassette = Cassette.load(path=file)
+    assert len(cassette) == 1

tests/integration/test_proxy.py

@@ -1,5 +1,6 @@
 """Test using a proxy."""
 
+import asyncio
 import http.server
 import socketserver
 import threading
@@ -36,6 +37,35 @@ class Proxy(http.server.SimpleHTTPRequestHandler):
         self.end_headers()
         self.copyfile(upstream_response, self.wfile)
 
+    def do_CONNECT(self):
+        host, port = self.path.split(":")
+
+        asyncio.run(self._tunnel(host, port, self.connection))
+
+    async def _tunnel(self, host, port, client_sock):
+        target_r, target_w = await asyncio.open_connection(host=host, port=port)
+
+        self.send_response(http.HTTPStatus.OK)
+        self.end_headers()
+
+        source_r, source_w = await asyncio.open_connection(sock=client_sock)
+
+        async def channel(reader, writer):
+            while True:
+                data = await reader.read(1024)
+                if not data:
+                    break
+                writer.write(data)
+                await writer.drain()
+
+            writer.close()
+            await writer.wait_closed()
+
+        await asyncio.gather(
+            channel(target_r, source_w),
+            channel(source_r, target_w),
+        )
+
 
 @pytest.fixture(scope="session")
 def proxy_server():
@@ -52,10 +82,26 @@ def test_use_proxy(tmpdir, httpbin, proxy_server):
     with vcr.use_cassette(str(tmpdir.join("proxy.yaml"))):
         response = requests.get(httpbin.url, proxies={"http": proxy_server})
 
-    with vcr.use_cassette(str(tmpdir.join("proxy.yaml")), mode="once") as cassette:
+    with vcr.use_cassette(str(tmpdir.join("proxy.yaml")), mode="none") as cassette:
         cassette_response = requests.get(httpbin.url, proxies={"http": proxy_server})
 
-    for key in set(cassette_response.headers.keys()) & set(response.headers.keys()):
-        assert cassette_response.headers[key] == response.headers[key]
     assert cassette_response.headers == response.headers
     assert cassette.play_count == 1
+
+
+def test_use_https_proxy(tmpdir, httpbin_secure, proxy_server):
+    """Ensure that it works with an HTTPS proxy."""
+    with vcr.use_cassette(str(tmpdir.join("proxy.yaml"))):
+        response = requests.get(httpbin_secure.url, proxies={"https": proxy_server})
+
+    with vcr.use_cassette(str(tmpdir.join("proxy.yaml")), mode="none") as cassette:
+        cassette_response = requests.get(
+            httpbin_secure.url,
+            proxies={"https": proxy_server},
+        )
+
+    assert cassette_response.headers == response.headers
+    assert cassette.play_count == 1
+
+    # The cassette URL points to httpbin, not the proxy
+    assert cassette.requests[0].url == httpbin_secure.url + "/"

tests/unit/test_cassettes.py

@@ -11,6 +11,7 @@ import yaml
 from vcr.cassette import Cassette
 from vcr.errors import UnhandledHTTPRequestError
 from vcr.patch import force_reset
+from vcr.request import Request
 from vcr.stubs import VCRHTTPSConnection
 
 
@@ -410,3 +411,25 @@ def test_find_requests_with_most_matches_many_similar_requests(mock_get_matchers
         (1, ["method", "path"], [("query", "failed : query")]),
         (3, ["method", "path"], [("query", "failed : query")]),
     ]
+
+
+def test_used_interactions(tmpdir):
+    interactions = [
+        {"request": {"body": "", "uri": "foo1", "method": "GET", "headers": {}}, "response": "bar1"},
+        {"request": {"body": "", "uri": "foo2", "method": "GET", "headers": {}}, "response": "bar2"},
+        {"request": {"body": "", "uri": "foo3", "method": "GET", "headers": {}}, "response": "bar3"},
+    ]
+    file = tmpdir.join("test_cassette.yml")
+    file.write(yaml.dump({"interactions": [interactions[0], interactions[1]]}))
+
+    cassette = Cassette.load(path=str(file))
+    request = Request._from_dict(interactions[1]["request"])
+    cassette.play_response(request)
+    assert len(cassette._played_interactions) < len(cassette._old_interactions)
+
+    request = Request._from_dict(interactions[2]["request"])
+    cassette.append(request, interactions[2]["response"])
+    assert len(cassette._new_interactions()) == 1
+
+    used_interactions = cassette._played_interactions + cassette._new_interactions()
+    assert len(used_interactions) == 2

vcr/cassette.py

@@ -3,7 +3,7 @@ import contextlib
 import copy
 import inspect
 import logging
-from asyncio import iscoroutinefunction
+from inspect import iscoroutinefunction
 
 import wrapt
 
@@ -177,6 +177,7 @@ class Cassette:
         custom_patches=(),
         inject=False,
         allow_playback_repeats=False,
+        drop_unused_requests=False,
     ):
         self._persister = persister or FilesystemPersister
         self._path = path
@@ -189,6 +190,7 @@ class Cassette:
         self.record_mode = record_mode
         self.custom_patches = custom_patches
         self.allow_playback_repeats = allow_playback_repeats
+        self.drop_unused_requests = drop_unused_requests
 
         # self.data is the list of (req, resp) tuples
         self.data = []
@@ -196,6 +198,10 @@ class Cassette:
         self.dirty = False
         self.rewound = False
 
+        # Subsets of self.data to store old and played interactions
+        self._old_interactions = []
+        self._played_interactions = []
+
     @property
     def play_count(self):
         return sum(self.play_counts.values())
@@ -257,6 +263,7 @@ class Cassette:
         for index, response in self._responses(request):
             if self.play_counts[index] == 0 or self.allow_playback_repeats:
                 self.play_counts[index] += 1
+                self._played_interactions.append((request, response))
                 return response
         # The cassette doesn't contain the request asked for.
         raise UnhandledHTTPRequestError(
@@ -317,12 +324,36 @@ class Cassette:
 
         return final_best_matches
 
+    def _new_interactions(self):
+        """List of new HTTP interactions (request/response tuples)"""
+        new_interactions = []
+        for request, response in self.data:
+            if all(
+                not requests_match(request, old_request, self._match_on)
+                for old_request, _ in self._old_interactions
+            ):
+                new_interactions.append((request, response))
+        return new_interactions
+
     def _as_dict(self):
         return {"requests": self.requests, "responses": self.responses}
 
+    def _build_used_interactions_dict(self):
+        interactions = self._played_interactions + self._new_interactions()
+        cassete_dict = {
+            "requests": [request for request, _ in interactions],
+            "responses": [response for _, response in interactions],
+        }
+        return cassete_dict
+
     def _save(self, force=False):
+        if self.drop_unused_requests and len(self._played_interactions) < len(self._old_interactions):
+            cassete_dict = self._build_used_interactions_dict()
+            force = True
+        else:
+            cassete_dict = self._as_dict()
         if force or self.dirty:
-            self._persister.save_cassette(self._path, self._as_dict(), serializer=self._serializer)
+            self._persister.save_cassette(self._path, cassete_dict, serializer=self._serializer)
             self.dirty = False
 
     def _load(self):
@@ -330,6 +361,7 @@ class Cassette:
             requests, responses = self._persister.load_cassette(self._path, serializer=self._serializer)
             for request, response in zip(requests, responses):
                 self.append(request, response)
+                self._old_interactions.append((request, response))
             self.dirty = False
             self.rewound = True
         except (CassetteDecodeError, CassetteNotFoundError):

vcr/config.py

@@ -48,6 +48,7 @@ class VCR:
         func_path_generator=None,
         decode_compressed_response=False,
         record_on_exception=True,
+        drop_unused_requests=False,
     ):
         self.serializer = serializer
         self.match_on = match_on
@@ -81,6 +82,7 @@ class VCR:
         self.decode_compressed_response = decode_compressed_response
         self.record_on_exception = record_on_exception
         self._custom_patches = tuple(custom_patches)
+        self.drop_unused_requests = drop_unused_requests
 
     def _get_serializer(self, serializer_name):
         try:
@@ -151,6 +153,7 @@ class VCR:
             "func_path_generator": func_path_generator,
             "allow_playback_repeats": kwargs.get("allow_playback_repeats", False),
             "record_on_exception": record_on_exception,
+            "drop_unused_requests": kwargs.get("drop_unused_requests", self.drop_unused_requests),
         }
         path = kwargs.get("path")
         if path:

vcr/request.py

@@ -27,6 +27,15 @@ class Request:
         self.headers = headers
         log.debug("Invoking Request %s", self.uri)
 
+    @property
+    def uri(self):
+        return self._uri
+
+    @uri.setter
+    def uri(self, uri):
+        self._uri = uri
+        self.parsed_uri = urlparse(uri)
+
     @property
     def headers(self):
         return self._headers
@@ -61,30 +70,29 @@ class Request:
 
     @property
     def scheme(self):
-        return urlparse(self.uri).scheme
+        return self.parsed_uri.scheme
 
     @property
     def host(self):
-        return urlparse(self.uri).hostname
+        return self.parsed_uri.hostname
 
     @property
     def port(self):
-        parse_uri = urlparse(self.uri)
-        port = parse_uri.port
+        port = self.parsed_uri.port
         if port is None:
             try:
-                port = {"https": 443, "http": 80}[parse_uri.scheme]
+                port = {"https": 443, "http": 80}[self.parsed_uri.scheme]
             except KeyError:
                 pass
         return port
 
     @property
     def path(self):
-        return urlparse(self.uri).path
+        return self.parsed_uri.path
 
     @property
     def query(self):
-        q = urlparse(self.uri).query
+        q = self.parsed_uri.query
         return sorted(parse_qsl(q))
 
     # alias for backwards compatibility

vcr/stubs/__init__.py

@@ -187,22 +187,34 @@ class VCRConnection:
         """
         Returns empty string for the default port and ':port' otherwise
         """
-        port = self.real_connection.port
+        port = (
+            self.real_connection.port
+            if not self.real_connection._tunnel_host
+            else self.real_connection._tunnel_port
+        )
         default_port = {"https": 443, "http": 80}[self._protocol]
         return f":{port}" if port != default_port else ""
 
+    def _real_host(self):
+        """Returns the request host"""
+        if self.real_connection._tunnel_host:
+            # The real connection is to an HTTPS proxy
+            return self.real_connection._tunnel_host
+        else:
+            return self.real_connection.host
+
     def _uri(self, url):
         """Returns request absolute URI"""
         if url and not url.startswith("/"):
             # Then this must be a proxy request.
             return url
-        uri = f"{self._protocol}://{self.real_connection.host}{self._port_postfix()}{url}"
+        uri = f"{self._protocol}://{self._real_host()}{self._port_postfix()}{url}"
         log.debug("Absolute URI: %s", uri)
         return uri
 
     def _url(self, uri):
         """Returns request selector url from absolute URI"""
-        prefix = f"{self._protocol}://{self.real_connection.host}{self._port_postfix()}"
+        prefix = f"{self._protocol}://{self._real_host()}{self._port_postfix()}"
         return uri.replace(prefix, "", 1)
 
     def request(self, method, url, body=None, headers=None, *args, **kwargs):