ci: build CPython 3.14 wheels (bump cibuildwheel to 4.x)

.github/workflows/python-package.yml

@@ -211,6 +211,42 @@ jobs:
       - name: Kafka broker logs (on failure)
         if: failure()
         run: docker logs --tail 40 kafka
+  test-redis-integration:
+    name: 'Redis cache integration'
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+    # Advisory for now, like the Kafka integration job: intentionally NOT in
+    # the `check` job's `needs`, so a red here does not block the merge queue.
+    continue-on-error: true
+    services:
+      redis:
+        image: redis:7
+        ports:
+          - 6379:6379
+        options: >-
+          --health-cmd "redis-cli ping"
+          --health-interval 5s
+          --health-timeout 3s
+          --health-retries 5
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          fetch-depth: 0
+      - uses: actions/setup-python@v5
+        with:
+          python-version: '3.12'
+          cache: pip
+          cache-dependency-path: requirements/test.txt
+      - name: Install dependencies
+        run: |
+          pip install -r requirements/test.txt
+          pip install .
+      # Runs the redis cache backend against the real server started above --
+      # the unit tests only ever mock the client.
+      - name: Run redis cache integration tests
+        env:
+          FAUST_TEST_REDIS: 'redis://localhost:6379'
+        run: pytest tests/integration/cache -v -ra --tb=short --no-cov
   check:  # This job does nothing and is only used for the branch protection
     name: ✅ Ensure the required checks passing
     if: always()
@@ -238,7 +274,10 @@ jobs:
         with:
           fetch-depth: 0
       - name: Build wheels
-        uses: pypa/cibuildwheel@v2.21.3
+        # cibuildwheel 2.21.3 predates CPython 3.14, so `build = "cp3*"`
+        # stopped at cp313 and no 3.14 wheels were published (issue #715).
+        # 4.x builds 3.14 by default and still supports cp310-cp313.
+        uses: pypa/cibuildwheel@v4.1.0
       - uses: actions/upload-artifact@v4
         with:
           name: cibw-wheels-${{ matrix.os }}

faust/web/cache/backends/redis.py

@@ -19,9 +19,12 @@
     import redis.asyncio as aredis
     import redis.exceptions
 
-    redis.client.Redis
+    redis.asyncio.client.Redis
 except ImportError:  # pragma: no cover
-    aredis = None  # noqa
+    # ``on_start`` (and the module-level ``if redis is None`` guards) key off
+    # ``redis`` being ``None`` when the library is missing; bind both names so
+    # the guard fires instead of raising ``NameError``.
+    redis = aredis = None  # noqa
 
 if typing.TYPE_CHECKING:  # pragma: no cover
     from redis import StrictRedis as _RedisClientT
@@ -84,12 +87,16 @@ def __init__(
         self._client_by_scheme = self._init_schemes()
 
     def _init_schemes(self) -> Mapping[str, Type[_RedisClientT]]:
-        if redis is None:  # pragma: no cover
+        if aredis is None:  # pragma: no cover
             return {}
         else:
+            # Use the asyncio clients: ``_get``/``_set``/``_delete`` await the
+            # client, so the synchronous ``redis.StrictRedis`` would raise
+            # "object ... can't be used in 'await' expression" against a real
+            # server (it only appeared to work because the tests mock it).
             return {
-                RedisScheme.SINGLE_NODE.value: redis.StrictRedis,
-                RedisScheme.CLUSTER.value: redis.RedisCluster,
+                RedisScheme.SINGLE_NODE.value: aredis.StrictRedis,
+                RedisScheme.CLUSTER.value: aredis.RedisCluster,
             }
 
     async def _get(self, key: str) -> Optional[bytes]:
@@ -117,6 +124,12 @@ async def on_start(self) -> None:
             )
         await self.connect()
 
+    async def on_stop(self) -> None:
+        """Call when Redis backend stops -- close the client connections."""
+        if self._client is not None:
+            await self._client.aclose()
+            self._client = None
+
     async def connect(self) -> None:
         """Connect to Redis/Redis Cluster server."""
         if self._client is None:

pyproject.toml

@@ -117,7 +117,12 @@ testpaths = [
 build = "cp3*"
 
 archs = ["auto64"]
-skip = ["*musllinux*"]
+# Skip musllinux, and skip the free-threaded builds (cp313t/cp314t): as of
+# cibuildwheel 3.1 free-threading is no longer experimental for 3.14, so
+# `cp3*` would otherwise build free-threaded wheels for the Cython extension,
+# which has not been validated under a no-GIL interpreter.  Drop `cp31?t-*`
+# from this list once faust is verified free-threading-safe.
+skip = ["*musllinux*", "cp31?t-*"]
 
 before-build = "pip install Cython"
 

tests/functional/conftest.py

@@ -127,10 +127,13 @@ def mocked_redis(*, event_loop, monkeypatch):
             setex=AsyncMock(side_effect=storage.setex),
             delete=AsyncMock(side_effect=storage.delete),
             ttl=AsyncMock(side_effect=storage.ttl),
+            aclose=AsyncMock(),
         ),
     )
     client_cls.storage = storage
-    monkeypatch.setattr("redis.StrictRedis", client_cls)
+    # The backend uses the asyncio client (redis.asyncio.StrictRedis), so patch
+    # that name rather than the synchronous redis.StrictRedis.
+    monkeypatch.setattr("redis.asyncio.StrictRedis", client_cls)
     return client_cls
 
 

tests/functional/web/test_cache.py

@@ -332,14 +332,17 @@ async def test_redis__using_starting_(app, mocked_redis):
 
 
 @pytest.fixture()
-def no_aredis(monkeypatch):
-    monkeypatch.setattr("faust.web.cache.backends.redis.aredis", None)
+def no_redis(monkeypatch):
+    # The backend's ``on_start`` guards on the ``redis`` module being
+    # importable (``if redis is None``); patch that name to ``None`` to
+    # simulate the library not being installed.  It must raise before any
+    # socket is opened.
+    monkeypatch.setattr("faust.web.cache.backends.redis.redis", None)
 
 
-@pytest.mark.skip(reason="Needs fixing")
 @pytest.mark.asyncio
 @pytest.mark.app(cache="redis://localhost:6079")
-async def test_redis__aredis_is_not_installed(*, app, no_aredis):
+async def test_redis__is_not_installed(*, app, no_redis):
     cache = app.cache
     with pytest.raises(ImproperlyConfigured):
         async with cache:
@@ -446,31 +449,45 @@ def bp(app):
     blueprint.register(app, url_prefix="/test/")
 
 
-@pytest.mark.skip(reason="Needs fixing")
 class Test_RedisScheme:
     def test_single_client(self, app):
         url = "redis://123.123.123.123:3636//1"
         Backend = backends.by_url(url)
         assert Backend is redis.CacheBackend
         backend = Backend(app, url=url)
         assert isinstance(backend, redis.CacheBackend)
+        # The single-node scheme maps to the asyncio StrictRedis client;
+        # constructing it does not open a socket (redis-py connects lazily).
+        assert (
+            backend._client_by_scheme[redis.RedisScheme.SINGLE_NODE.value]
+            is aredis.StrictRedis
+        )
         client = backend._new_client()
-        assert isinstance(client, redis.StrictRedis)
+        assert isinstance(client, aredis.StrictRedis)
         pool = client.connection_pool
         assert pool.connection_kwargs["host"] == backend.url.host
         assert pool.connection_kwargs["port"] == backend.url.port
         assert pool.connection_kwargs["db"] == 1
 
-    def test_cluster_client(self, app):
+    def test_cluster_client(self, app, monkeypatch):
+        # A real RedisCluster client connects on construction, so stand in the
+        # (external) asyncio cluster class with a mock -- mirroring the
+        # ``mocked_redis`` fixture that patches the single-node client.
+        cluster_cls = Mock(name="RedisCluster")
+        monkeypatch.setattr("redis.asyncio.RedisCluster", cluster_cls)
+
         url = "rediscluster://123.123.123.123:3636//1"
         Backend = backends.by_url(url)
         assert Backend is redis.CacheBackend
         backend = Backend(app, url=url)
         assert isinstance(backend, redis.CacheBackend)
+        assert backend._client_by_scheme[redis.RedisScheme.CLUSTER.value] is cluster_cls
         client = backend._new_client()
-        assert isinstance(client, aredis.RedisCluster)
-        pool = client.connection_pool
-        assert {
-            "host": backend.url.host,
-            "port": 3636,
-        } in pool.nodes.startup_nodes
+        assert client is cluster_cls.return_value
+        cluster_cls.assert_called_once()
+        _, kwargs = cluster_cls.call_args
+        assert kwargs["host"] == backend.url.host
+        assert kwargs["port"] == backend.url.port
+        # Redis Cluster does not accept a ``db`` argument, so the backend
+        # must strip it (see CacheBackend._as_cluster_kwargs).
+        assert "db" not in kwargs

tests/integration/cache/conftest.py

@@ -0,0 +1,18 @@
+"""Fixtures for live-server cache integration tests."""
+
+import asyncio
+
+import pytest
+
+
+@pytest.fixture(autouse=True)
+async def _shutdown_default_executor():
+    """Reap the loop's default executor thread after each test.
+
+    redis-py's asyncio client resolves the server address through the event
+    loop's default :class:`~concurrent.futures.ThreadPoolExecutor`, which
+    otherwise lingers as an ``asyncio_N`` thread and trips the strict
+    ``threads_not_lingering`` guard in ``tests/conftest.py``.
+    """
+    yield
+    await asyncio.get_event_loop().shutdown_default_executor()

tests/integration/cache/test_redis.py

@@ -0,0 +1,73 @@
+"""Live-server integration tests for the Redis cache backend.
+
+These exercise :mod:`faust.web.cache.backends.redis` against a *real* Redis
+server -- the unit tests mock the client, which is exactly why the backend
+could ship awaiting a synchronous client without anyone noticing.  Point
+``FAUST_TEST_REDIS`` at a reachable server (default ``redis://localhost:6379``)
+to run them; they skip when no server is reachable.
+"""
+
+import os
+import socket
+
+import pytest
+from yarl import URL
+
+import faust
+
+REDIS_URL = os.environ.get("FAUST_TEST_REDIS", "redis://localhost:6379")
+
+
+def _redis_reachable(url: str) -> bool:
+    u = URL(url)
+    try:
+        with socket.create_connection((u.host or "localhost", u.port or 6379), 2):
+            return True
+    except OSError:
+        return False
+
+
+pytestmark = [
+    pytest.mark.asyncio,
+    # Live sockets can emit ResourceWarning during teardown; the suite escalates
+    # those to errors (pyproject.toml), so relax it here as the broker tests do.
+    pytest.mark.filterwarnings("ignore::ResourceWarning"),
+    pytest.mark.skipif(
+        not _redis_reachable(REDIS_URL),
+        reason=f"no reachable Redis at {REDIS_URL} (set FAUST_TEST_REDIS)",
+    ),
+]
+
+
+def _make_app() -> faust.App:
+    return faust.App(
+        "faust-integration-redis-cache",
+        cache=REDIS_URL,
+        # a unique key prefix per run keeps parallel/repeated runs isolated
+        broker="memory://",
+    )
+
+
+async def test_redis_cache_set_get_delete():
+    app = _make_app()
+    async with app.cache:
+        key = "faust-it:roundtrip"
+        await app.cache.delete(key)
+        assert await app.cache.get(key) is None
+
+        await app.cache.set(key, b"value-1", timeout=60)
+        assert await app.cache.get(key) == b"value-1"
+
+        # overwrite
+        await app.cache.set(key, b"value-2", timeout=60)
+        assert await app.cache.get(key) == b"value-2"
+
+        # delete
+        await app.cache.delete(key)
+        assert await app.cache.get(key) is None
+
+
+async def test_redis_cache_missing_key_returns_none():
+    app = _make_app()
+    async with app.cache:
+        assert await app.cache.get("faust-it:definitely-not-set") is None