diff --git a/.github/workflows/python-package.yml b/.github/workflows/python-package.yml index 161e4293d..e740a6136 100644 --- a/.github/workflows/python-package.yml +++ b/.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 }} diff --git a/faust/web/cache/backends/redis.py b/faust/web/cache/backends/redis.py index 7661e5aba..13773523b 100644 --- a/faust/web/cache/backends/redis.py +++ b/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: diff --git a/pyproject.toml b/pyproject.toml index aaf28974a..f4729860e 100644 --- a/pyproject.toml +++ b/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" diff --git a/tests/functional/conftest.py b/tests/functional/conftest.py index ae3300153..4dc8b8182 100644 --- a/tests/functional/conftest.py +++ b/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 diff --git a/tests/functional/web/test_cache.py b/tests/functional/web/test_cache.py index 880a9a2d8..305d7df8c 100644 --- a/tests/functional/web/test_cache.py +++ b/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,7 +449,6 @@ 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" @@ -454,23 +456,38 @@ def test_single_client(self, app): 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 diff --git a/tests/integration/cache/__init__.py b/tests/integration/cache/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/integration/cache/conftest.py b/tests/integration/cache/conftest.py new file mode 100644 index 000000000..091a3de97 --- /dev/null +++ b/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() diff --git a/tests/integration/cache/test_redis.py b/tests/integration/cache/test_redis.py new file mode 100644 index 000000000..08feef8a5 --- /dev/null +++ b/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