diff --git a/ldclient/async_config.py b/ldclient/async_config.py new file mode 100644 index 00000000..9a37a133 --- /dev/null +++ b/ldclient/async_config.py @@ -0,0 +1,65 @@ +""" +Configuration types for the async SDK client. Currently this holds +:class:`AsyncBigSegmentsConfig`; the full :class:`AsyncConfig` lands in a later slice. + +.. caution:: + This feature is experimental and should NOT be considered ready for production + use. It may change or be removed without notice and is not subject to backwards + compatibility guarantees. +""" + +from typing import Optional + +from ldclient.interfaces import AsyncBigSegmentStore + + +class AsyncBigSegmentsConfig: + """Configuration options related to Big Segments for the async SDK client. + + Big Segments are a specific type of segments. For more information, read the LaunchDarkly + documentation: https://docs.launchdarkly.com/home/users/big-segments + + If your application uses Big Segments, you will need to create an ``AsyncBigSegmentsConfig`` + that at a minimum specifies what database integration to use, and then pass the + ``AsyncBigSegmentsConfig`` object as the ``big_segments`` parameter when creating an + :class:`AsyncConfig`. + """ + + def __init__(self, store: Optional[AsyncBigSegmentStore] = None, context_cache_size: int = 1000, context_cache_time: float = 5, status_poll_interval: float = 5, stale_after: float = 120): + """ + :param store: the implementation of :class:`ldclient.interfaces.AsyncBigSegmentStore` that + will be used to query the Big Segments database + :param context_cache_size: the maximum number of contexts whose Big Segment state will be cached + by the SDK at any given time + :param context_cache_time: the maximum length of time (in seconds) that the Big Segment state + for a context will be cached by the SDK + :param status_poll_interval: the interval (in seconds) at which the SDK will poll the Big + Segment store to make sure it is available and to determine how long ago it was updated + :param stale_after: the maximum length of time between updates of the Big Segments data + before the data is considered out of date + """ + self.__store = store + self.__context_cache_size = context_cache_size + self.__context_cache_time = context_cache_time + self.__status_poll_interval = status_poll_interval + self.__stale_after = stale_after + + @property + def store(self) -> Optional[AsyncBigSegmentStore]: + return self.__store + + @property + def context_cache_size(self) -> int: + return self.__context_cache_size + + @property + def context_cache_time(self) -> float: + return self.__context_cache_time + + @property + def status_poll_interval(self) -> float: + return self.__status_poll_interval + + @property + def stale_after(self) -> float: + return self.__stale_after diff --git a/ldclient/impl/async_big_segments.py b/ldclient/impl/async_big_segments.py new file mode 100644 index 00000000..a35301fc --- /dev/null +++ b/ldclient/impl/async_big_segments.py @@ -0,0 +1,98 @@ +from typing import Optional, Tuple + +from expiringdict import ExpiringDict + +from ldclient.async_config import AsyncBigSegmentsConfig +from ldclient.evaluation import BigSegmentsStatus +from ldclient.impl.aio.concurrency import AsyncRepeatingTask +from ldclient.impl.big_segments_common import ( + EMPTY_MEMBERSHIP, + BigSegmentStoreStatusProviderImpl, + _hash_for_user_key, + is_stale +) +from ldclient.impl.util import log +from ldclient.interfaces import ( + BigSegmentStoreStatus, + BigSegmentStoreStatusProvider +) + + +class AsyncBigSegmentStoreManager: + """ + Internal component that decorates the Big Segment store with caching behavior, and also polls the + store to track its status. The constructor starts the polling task. + """ + + # Because the constructor starts the polling task, it must run within a running event loop. + def __init__(self, config: AsyncBigSegmentsConfig): + self.__store = config.store + + self.__stale_after_millis = config.stale_after * 1000 + self.__status_provider = BigSegmentStoreStatusProviderImpl(self.get_status) + self.__last_status = None # type: Optional[BigSegmentStoreStatus] + self.__poll_task = None # type: Optional[AsyncRepeatingTask] + + if self.__store: + self.__cache = ExpiringDict(max_len=config.context_cache_size, max_age_seconds=config.context_cache_time) + self.__poll_task = AsyncRepeatingTask("ldclient.bigsegment.status-poll", config.status_poll_interval, 0, self.poll_store_and_update_status) + self.__poll_task.start() + + async def stop(self): + if self.__poll_task: + self.__poll_task.stop() + if self.__store: + await self.__store.stop() + + @property + def status_provider(self) -> BigSegmentStoreStatusProvider: + return self.__status_provider + + async def get_user_membership(self, user_key: str) -> Tuple[Optional[dict], str]: + if not self.__store: + return None, BigSegmentsStatus.NOT_CONFIGURED + membership = self.__cache.get(user_key) + if membership is None: + user_hash = _hash_for_user_key(user_key) + try: + membership = await self.__store.get_membership(user_hash) # type: ignore[misc] + if membership is None: + membership = EMPTY_MEMBERSHIP + self.__cache[user_key] = membership + except Exception as e: + log.exception("Big Segment store membership query returned error: %s" % e) + return None, BigSegmentsStatus.STORE_ERROR + # First-call fallback: if the polling task hasn't run yet, poll inline now + status = self.__last_status + if status is None: + status = await self.poll_store_and_update_status() + if not status.available: + return membership, BigSegmentsStatus.STORE_ERROR + return membership, BigSegmentsStatus.STALE if status.stale else BigSegmentsStatus.HEALTHY + + def get_status(self) -> BigSegmentStoreStatus: + """Return the most recently polled status. + + When no status has been cached yet, the sync variant polls the store + inline; the async variant (whose status getter cannot await) reports + the store as unavailable until the polling task has run. + """ + status = self.__last_status + if status is None: + return BigSegmentStoreStatus(False, False) + return status if status else self.poll_store_and_update_status() # type: ignore[return-value] + + async def poll_store_and_update_status(self) -> BigSegmentStoreStatus: + new_status = BigSegmentStoreStatus(False, False) # default to "unavailable" if we don't get a new status below + if self.__store: + try: + metadata = await self.__store.get_metadata() # type: ignore[misc] + new_status = BigSegmentStoreStatus(True, (metadata is None) or self.is_stale(metadata.last_up_to_date)) + except Exception as e: + log.exception("Big Segment store status query returned error: %s" % e) + self.__last_status = new_status + self.__status_provider._update_status(new_status) + return new_status + + def is_stale(self, timestamp) -> bool: + return is_stale(timestamp, self.__stale_after_millis) diff --git a/ldclient/impl/big_segments.py b/ldclient/impl/big_segments.py index b96c5ef6..0f42c07d 100644 --- a/ldclient/impl/big_segments.py +++ b/ldclient/impl/big_segments.py @@ -1,13 +1,15 @@ -import base64 -import time -from hashlib import sha256 -from typing import Callable, Optional, Tuple +from typing import Optional, Tuple from expiringdict import ExpiringDict from ldclient.config import BigSegmentsConfig from ldclient.evaluation import BigSegmentsStatus -from ldclient.impl.listeners import Listeners +from ldclient.impl.big_segments_common import ( + EMPTY_MEMBERSHIP, + BigSegmentStoreStatusProviderImpl, + _hash_for_user_key, + is_stale +) from ldclient.impl.repeating_task import RepeatingTask from ldclient.impl.util import log from ldclient.interfaces import ( @@ -16,44 +18,7 @@ ) -class BigSegmentStoreStatusProviderImpl(BigSegmentStoreStatusProvider): - """ - Default implementation of the BigSegmentStoreStatusProvider interface. - - The real implementation of getting the status is in BigSegmentStoreManager - we pass in a lambda that - allows us to get the current status from that class. So this class provides a facade for that, and - also adds the listener mechanism. - """ - - def __init__(self, status_getter: Callable[[], BigSegmentStoreStatus]): - self.__status_getter = status_getter - self.__status_listeners = Listeners() - self.__last_status = None # type: Optional[BigSegmentStoreStatus] - - @property - def status(self) -> BigSegmentStoreStatus: - return self.__status_getter() - - def add_listener(self, listener: Callable[[BigSegmentStoreStatus], None]) -> None: - self.__status_listeners.add(listener) - - def remove_listener(self, listener: Callable[[BigSegmentStoreStatus], None]) -> None: - self.__status_listeners.remove(listener) - - def _update_status(self, new_status: BigSegmentStoreStatus): - last = self.__last_status - if last is None: - self.__last_status = new_status - elif new_status.available != last.available or new_status.stale != last.stale: - self.__last_status = new_status - self.__status_listeners.notify(new_status) - - class BigSegmentStoreManager: - # use EMPTY_MEMBERSHIP as a singleton whenever a membership query returns None; it's safe to reuse it - # because we will never modify the membership properties after they're queried - EMPTY_MEMBERSHIP = {} # type: dict - """ Internal component that decorates the Big Segment store with caching behavior, and also polls the store to track its status. @@ -91,7 +56,7 @@ def get_user_membership(self, user_key: str) -> Tuple[Optional[dict], str]: try: membership = self.__store.get_membership(user_hash) if membership is None: - membership = self.EMPTY_MEMBERSHIP + membership = EMPTY_MEMBERSHIP self.__cache[user_key] = membership except Exception as e: log.exception("Big Segment store membership query returned error: %s" % e) @@ -120,8 +85,4 @@ def poll_store_and_update_status(self) -> BigSegmentStoreStatus: return new_status def is_stale(self, timestamp) -> bool: - return (timestamp is None) or ((int(time.time() * 1000) - timestamp) >= self.__stale_after_millis) - - -def _hash_for_user_key(user_key: str) -> str: - return base64.b64encode(sha256(user_key.encode('utf-8')).digest()).decode('utf-8') + return is_stale(timestamp, self.__stale_after_millis) diff --git a/ldclient/impl/big_segments_common.py b/ldclient/impl/big_segments_common.py new file mode 100644 index 00000000..08f4f62e --- /dev/null +++ b/ldclient/impl/big_segments_common.py @@ -0,0 +1,63 @@ +""" +Shared, I/O-free big-segments status provider used by both the sync +:mod:`ldclient.impl.big_segments` and the async +:mod:`ldclient.impl.async_big_segments`. It holds the last known status and +notifies listeners; nothing here touches the store or network, so it is +identical across the two managers. +""" + +import base64 +import time +from hashlib import sha256 +from typing import Callable, Optional + +from ldclient.impl.listeners import Listeners +from ldclient.interfaces import ( + BigSegmentStoreStatus, + BigSegmentStoreStatusProvider +) + +# use EMPTY_MEMBERSHIP as a singleton whenever a membership query returns None; it's safe to reuse it +# because we will never modify the membership properties after they're queried +EMPTY_MEMBERSHIP = {} # type: dict + + +def _hash_for_user_key(user_key: str) -> str: + return base64.b64encode(sha256(user_key.encode('utf-8')).digest()).decode('utf-8') + + +def is_stale(timestamp: int, stale_after_millis) -> bool: + return (timestamp is None) or ((int(time.time() * 1000) - timestamp) >= stale_after_millis) + + +class BigSegmentStoreStatusProviderImpl(BigSegmentStoreStatusProvider): + """ + Default implementation of the BigSegmentStoreStatusProvider interface. + + The real implementation of getting the status is in the big segment store manager - we pass in a lambda that + allows us to get the current status from that class. So this class provides a facade for that, and + also adds the listener mechanism. + """ + + def __init__(self, status_getter: Callable[[], BigSegmentStoreStatus]): + self.__status_getter = status_getter + self.__status_listeners = Listeners() + self.__last_status = None # type: Optional[BigSegmentStoreStatus] + + @property + def status(self) -> BigSegmentStoreStatus: + return self.__status_getter() + + def add_listener(self, listener: Callable[[BigSegmentStoreStatus], None]) -> None: + self.__status_listeners.add(listener) + + def remove_listener(self, listener: Callable[[BigSegmentStoreStatus], None]) -> None: + self.__status_listeners.remove(listener) + + def _update_status(self, new_status: BigSegmentStoreStatus): + last = self.__last_status + if last is None: + self.__last_status = new_status + elif new_status.available != last.available or new_status.stale != last.stale: + self.__last_status = new_status + self.__status_listeners.notify(new_status) diff --git a/ldclient/impl/integrations/redis/async_redis_big_segment_store.py b/ldclient/impl/integrations/redis/async_redis_big_segment_store.py new file mode 100644 index 00000000..7eebb6f3 --- /dev/null +++ b/ldclient/impl/integrations/redis/async_redis_big_segment_store.py @@ -0,0 +1,41 @@ +from typing import Any, Dict, Optional + +from ldclient.impl.util import log, redact_password +from ldclient.interfaces import AsyncBigSegmentStore, BigSegmentStoreMetadata + +have_async_redis = False +try: + import redis.asyncio as redis_client + + have_async_redis = True +except ImportError: + pass + + +class _AsyncRedisBigSegmentStore(AsyncBigSegmentStore): + KEY_LAST_UP_TO_DATE = ':big_segments_synchronized_on' + KEY_USER_INCLUDE = ':big_segment_include:' + KEY_USER_EXCLUDE = ':big_segment_exclude:' + + def __init__(self, url: str, prefix: Optional[str], redis_opts: Dict[str, Any]): + if not have_async_redis: + raise NotImplementedError("Cannot use async Redis Big Segment store because redis package is not installed") + self._prefix = prefix or 'launchdarkly' + self._client = redis_client.from_url(url, **redis_opts) + log.info("Started AsyncRedisBigSegmentStore connected to URL: " + redact_password(url) + " using prefix: " + self._prefix) + + async def get_metadata(self) -> BigSegmentStoreMetadata: + value = await self._client.get(self._prefix + self.KEY_LAST_UP_TO_DATE) + return BigSegmentStoreMetadata(int(value) if value else None) + + async def get_membership(self, user_hash: str) -> Optional[dict]: + included = await self._client.smembers(self._prefix + self.KEY_USER_INCLUDE + user_hash) + excluded = await self._client.smembers(self._prefix + self.KEY_USER_EXCLUDE + user_hash) + if not included and not excluded: + return None + ret = {ref.decode(): False for ref in excluded} + ret.update({ref.decode(): True for ref in included}) + return ret + + async def stop(self): + await self._client.aclose() diff --git a/ldclient/integrations/__init__.py b/ldclient/integrations/__init__.py index e1a560ce..a16347ca 100644 --- a/ldclient/integrations/__init__.py +++ b/ldclient/integrations/__init__.py @@ -237,6 +237,40 @@ def new_big_segment_store(url: str = 'redis://localhost:6379/0', prefix: str = ' return _RedisBigSegmentStore(url, prefix, redis_opts) + @staticmethod + def async_big_segment_store(url: str = 'redis://localhost:6379/0', prefix: Optional[str] = None, redis_opts: Dict[str, Any] = {}): + """ + Creates an async Redis-backed Big Segment store implementing :class:`~ldclient.interfaces.AsyncBigSegmentStore`. + + .. caution:: + This feature is experimental and should NOT be considered ready for production + use. It may change or be removed without notice and is not subject to backwards + compatibility guarantees. Pin to a specific minor version and review the changelog + before upgrading. + + Big Segments are a specific type of user segments. For more information, read the LaunchDarkly + documentation: https://docs.launchdarkly.com/home/users/big-segments + + To use this method, you must first install the ``redis`` package (version >=5.0.1). Then, put + the object returned by this method into the ``store`` property of your Big Segments + configuration when constructing an ``AsyncLDClient``. + :: + + from ldclient.config import Config, BigSegmentsConfig + from ldclient.integrations import Redis + store = Redis.async_big_segment_store() + config = Config(big_segments=BigSegmentsConfig(store=store)) + + :param url: the URL of the Redis host; defaults to ``DEFAULT_URL`` + :param prefix: a namespace prefix to be prepended to all Redis keys; defaults to + ``DEFAULT_PREFIX`` + :param redis_opts: extra options forwarded to ``redis.asyncio.from_url`` + """ + from ldclient.impl.integrations.redis.async_redis_big_segment_store import ( + _AsyncRedisBigSegmentStore + ) + return _AsyncRedisBigSegmentStore(url, prefix, redis_opts) + class Files: """Provides factory methods for integrations with filesystem data.""" diff --git a/ldclient/testing/impl/test_async_big_segments.py b/ldclient/testing/impl/test_async_big_segments.py new file mode 100644 index 00000000..fe83ea29 --- /dev/null +++ b/ldclient/testing/impl/test_async_big_segments.py @@ -0,0 +1,328 @@ +""" +Unit tests for AsyncBigSegmentStoreManager. + +These tests use a mock AsyncBigSegmentStore and require no real Redis connection. +""" + +import asyncio +import time + +import pytest + +from ldclient.async_config import AsyncBigSegmentsConfig +from ldclient.evaluation import BigSegmentsStatus +from ldclient.impl.async_big_segments import ( + AsyncBigSegmentStoreManager, + _hash_for_user_key +) +from ldclient.interfaces import AsyncBigSegmentStore, BigSegmentStoreMetadata + +# --------------------------------------------------------------------------- +# Test doubles +# --------------------------------------------------------------------------- + +user_key = 'user-key' +user_hash = _hash_for_user_key(user_key) + + +class MockAsyncBigSegmentStore(AsyncBigSegmentStore): + """In-process async mock that records calls and allows controlled responses.""" + + def __init__(self): + self._membership_queries = [] + self._memberships = {} + self._metadata_fn = lambda: BigSegmentStoreMetadata(int(time.time() * 1000)) + self._stopped = False + + def setup_membership(self, user_hash: str, membership): + self._memberships[user_hash] = membership + + def setup_metadata_always_up_to_date(self): + self._metadata_fn = lambda: BigSegmentStoreMetadata(int(time.time() * 1000)) + + def setup_metadata_always_stale(self): + self._metadata_fn = lambda: BigSegmentStoreMetadata(0) + + def setup_metadata_none(self): + self._metadata_fn = lambda: None + + def setup_metadata_error(self): + def _raise(): + raise Exception("deliberate metadata error") + self._metadata_fn = _raise + + async def get_metadata(self) -> BigSegmentStoreMetadata: + return self._metadata_fn() + + async def get_membership(self, context_hash: str): + self._membership_queries.append(context_hash) + return self._memberships.get(context_hash, None) + + async def stop(self): + self._stopped = True + + @property + def membership_queries(self): + return list(self._membership_queries) + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +async def make_started_manager(store, **kwargs): + config = AsyncBigSegmentsConfig(store=store, **kwargs) + # The constructor starts the polling task (it requires a running event loop). + return AsyncBigSegmentStoreManager(config) + + +# --------------------------------------------------------------------------- +# Tests +# --------------------------------------------------------------------------- + +@pytest.mark.asyncio +async def test_membership_query_uncached_result_healthy_status(): + store = MockAsyncBigSegmentStore() + store.setup_metadata_always_up_to_date() + expected_membership = {"key1": True, "key2": False} + store.setup_membership(user_hash, expected_membership) + + manager = await make_started_manager(store) + try: + membership, status = await manager.get_user_membership(user_key) + assert membership == expected_membership + assert status == BigSegmentsStatus.HEALTHY + finally: + await manager.stop() + + +@pytest.mark.asyncio +async def test_membership_query_cached_result_healthy_status(): + store = MockAsyncBigSegmentStore() + store.setup_metadata_always_up_to_date() + expected_membership = {"key1": True, "key2": False} + store.setup_membership(user_hash, expected_membership) + + manager = await make_started_manager(store) + try: + result1 = await manager.get_user_membership(user_key) + result2 = await manager.get_user_membership(user_key) + assert result1 == result2 + finally: + await manager.stop() + + # Only one query to the store despite two calls — cache hit on second call + assert store.membership_queries == [user_hash] + + +@pytest.mark.asyncio +async def test_membership_query_can_cache_result_of_none(): + store = MockAsyncBigSegmentStore() + store.setup_metadata_always_up_to_date() + store.setup_membership(user_hash, None) # store returns None → EMPTY_MEMBERSHIP cached + + manager = await make_started_manager(store) + try: + membership1, status1 = await manager.get_user_membership(user_key) + membership2, status2 = await manager.get_user_membership(user_key) + # None from store becomes EMPTY_MEMBERSHIP ({}) + assert membership1 == {} + assert membership2 == {} + assert status1 == BigSegmentsStatus.HEALTHY + assert status2 == BigSegmentsStatus.HEALTHY + finally: + await manager.stop() + + assert store.membership_queries == [user_hash] # only 1 query, cache hit on second + + +@pytest.mark.asyncio +async def test_membership_query_stale_status(): + store = MockAsyncBigSegmentStore() + store.setup_metadata_always_stale() + expected_membership = {"key1": True, "key2": False} + store.setup_membership(user_hash, expected_membership) + + manager = await make_started_manager(store) + try: + membership, status = await manager.get_user_membership(user_key) + assert membership == expected_membership + assert status == BigSegmentsStatus.STALE + finally: + await manager.stop() + + +@pytest.mark.asyncio +async def test_membership_query_stale_status_no_store_metadata(): + store = MockAsyncBigSegmentStore() + store.setup_metadata_none() + expected_membership = {"key1": True, "key2": False} + store.setup_membership(user_hash, expected_membership) + + manager = await make_started_manager(store) + try: + membership, status = await manager.get_user_membership(user_key) + assert membership == expected_membership + assert status == BigSegmentsStatus.STALE + finally: + await manager.stop() + + +@pytest.mark.asyncio +async def test_membership_query_store_error_returns_store_error_status(): + store = MockAsyncBigSegmentStore() + store.setup_metadata_always_up_to_date() + # No membership set up — store will raise on get_membership via a broken store + broken_store = MockAsyncBigSegmentStore() + broken_store.setup_metadata_always_up_to_date() + + async def _raise_on_membership(context_hash): + raise Exception("deliberate membership error") + + broken_store.get_membership = _raise_on_membership + + manager = await make_started_manager(broken_store) + try: + membership, status = await manager.get_user_membership(user_key) + assert membership is None + assert status == BigSegmentsStatus.STORE_ERROR + finally: + await manager.stop() + + +@pytest.mark.asyncio +async def test_no_store_returns_not_configured(): + config = AsyncBigSegmentsConfig(store=None) + manager = AsyncBigSegmentStoreManager(config) + try: + membership, status = await manager.get_user_membership(user_key) + assert membership is None + assert status == BigSegmentsStatus.NOT_CONFIGURED + finally: + await manager.stop() + + +@pytest.mark.asyncio +async def test_first_call_before_poll_task_triggers_inline_poll(): + """ + When get_user_membership is called before the background polling task has run, + poll_store_and_update_status() must be called inline so that a valid status is returned + rather than deterministically returning STORE_ERROR. + """ + store = MockAsyncBigSegmentStore() + store.setup_metadata_always_up_to_date() + store.setup_membership(user_hash, {"seg1": True}) + + # Query immediately after construction, racing the just-started poll task; + # whichever runs first, a valid status must be returned. + config = AsyncBigSegmentsConfig(store=store) + manager = AsyncBigSegmentStoreManager(config) + try: + membership, status = await manager.get_user_membership(user_key) + # Should have done an inline poll and returned HEALTHY, not STORE_ERROR + assert status == BigSegmentsStatus.HEALTHY + assert membership == {"seg1": True} + finally: + await manager.stop() + + +@pytest.mark.asyncio +async def test_concurrent_calls_for_same_key_all_return_correct_result(): + """ + Multiple concurrent coroutines missing the cache for the same key may each fetch from + the store independently, but all must return the same correct result. Duplicate fetches + are harmless because they return identical data and both populate the cache. + """ + original_store = MockAsyncBigSegmentStore() + original_store.setup_metadata_always_up_to_date() + original_store.setup_membership(user_hash, {"seg1": True}) + + class SlowStore(MockAsyncBigSegmentStore): + async def get_membership(self, context_hash): + await asyncio.sleep(0.05) # introduce latency to make concurrency visible + return await super().get_membership(context_hash) + + slow_store = SlowStore() + slow_store.setup_metadata_always_up_to_date() + slow_store.setup_membership(user_hash, {"seg1": True}) + + manager = await make_started_manager(slow_store) + try: + # Fire 5 concurrent requests for the same key before any has populated the cache + results = await asyncio.gather(*[manager.get_user_membership(user_key) for _ in range(5)]) + finally: + await manager.stop() + + # All results should be identical and correct + assert all(r == results[0] for r in results) + assert results[0] == ({"seg1": True}, BigSegmentsStatus.HEALTHY) + + +@pytest.mark.asyncio +async def test_status_reflects_store_unavailable(): + store = MockAsyncBigSegmentStore() + store.setup_metadata_error() + + manager = await make_started_manager(store) + try: + # Force a poll + status = await manager.poll_store_and_update_status() + assert status.available is False + finally: + await manager.stop() + + +@pytest.mark.asyncio +async def test_status_reflects_stale(): + store = MockAsyncBigSegmentStore() + store.setup_metadata_always_stale() + + manager = await make_started_manager(store) + try: + status = await manager.poll_store_and_update_status() + assert status.available is True + assert status.stale is True + finally: + await manager.stop() + + +@pytest.mark.asyncio +async def test_status_reflects_healthy(): + store = MockAsyncBigSegmentStore() + store.setup_metadata_always_up_to_date() + + manager = await make_started_manager(store) + try: + status = await manager.poll_store_and_update_status() + assert status.available is True + assert status.stale is False + finally: + await manager.stop() + + +@pytest.mark.asyncio +async def test_stop_stops_store(): + store = MockAsyncBigSegmentStore() + store.setup_metadata_always_up_to_date() + + manager = await make_started_manager(store) + await manager.stop() + + assert store._stopped is True + + +@pytest.mark.asyncio +async def test_status_provider_status_is_synchronous(): + """BigSegmentStoreStatusProvider.status must be readable synchronously without an await.""" + store = MockAsyncBigSegmentStore() + store.setup_metadata_always_up_to_date() + + manager = await make_started_manager(store) + try: + # Poll once to populate __last_status + await manager.poll_store_and_update_status() + # status property is sync — calling it should not raise + status = manager.status_provider.status + assert status.available is True + finally: + await manager.stop() diff --git a/ldclient/testing/integrations/test_async_redis.py b/ldclient/testing/integrations/test_async_redis.py new file mode 100644 index 00000000..3ebb88f3 --- /dev/null +++ b/ldclient/testing/integrations/test_async_redis.py @@ -0,0 +1,202 @@ +""" +Integration tests for _AsyncRedisBigSegmentStore. + +These tests require a real Redis instance running on localhost:6379. +They are skipped when the redis package is not installed or when the +LD_SKIP_DATABASE_TESTS environment variable is set to '1'. +""" + +from os import environ +from typing import List + +import pytest + +from ldclient.interfaces import BigSegmentStoreMetadata +from ldclient.testing.test_util import skip_database_tests + +have_async_redis = False +try: + import redis.asyncio as aioredis + + have_async_redis = True +except ImportError: + pass + +try: + import redis as _sync_redis + + have_sync_redis = True +except ImportError: + have_sync_redis = False + +# Skip marker applied to the whole module +pytestmark = pytest.mark.skipif( + not have_async_redis, + reason="skipping async Redis tests because redis package (>=4.6) is not installed" +) + +DEFAULT_PREFIX = 'launchdarkly' +FAKE_USER_HASH = 'userhash' + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def sync_redis_client(): + """Return a synchronous Redis client for test setup/teardown.""" + import redis + return redis.StrictRedis(host="localhost", port=6379, db=0) + + +def clear_data(prefix: str): + r = sync_redis_client() + pattern = "%s:*" % (prefix or DEFAULT_PREFIX) + for key in r.keys(pattern): + r.delete(key) + + +def set_metadata(prefix: str, metadata: BigSegmentStoreMetadata): + from ldclient.impl.integrations.redis.async_redis_big_segment_store import ( + _AsyncRedisBigSegmentStore + ) + r = sync_redis_client() + key = (prefix or DEFAULT_PREFIX) + _AsyncRedisBigSegmentStore.KEY_LAST_UP_TO_DATE + if metadata.last_up_to_date is None: + r.set(key, "") + else: + r.set(key, str(metadata.last_up_to_date)) + + +def set_segments(prefix: str, user_hash: str, includes: List[str], excludes: List[str]): + from ldclient.impl.integrations.redis.async_redis_big_segment_store import ( + _AsyncRedisBigSegmentStore + ) + r = sync_redis_client() + pfx = prefix or DEFAULT_PREFIX + for ref in includes: + r.sadd(pfx + _AsyncRedisBigSegmentStore.KEY_USER_INCLUDE + user_hash, ref) + for ref in excludes: + r.sadd(pfx + _AsyncRedisBigSegmentStore.KEY_USER_EXCLUDE + user_hash, ref) + + +def make_store(prefix=None): + from ldclient.integrations import Redis + return Redis.async_big_segment_store(prefix=prefix) + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + +@pytest.fixture(params=[None, 'testprefix']) +def prefix(request): + return request.param + + +@pytest.fixture(autouse=True) +def clear_before_each(prefix): + if skip_database_tests or not have_sync_redis: + yield + return + try: + clear_data(prefix) + except Exception: + pass + yield + try: + clear_data(prefix) + except Exception: + pass + + +# --------------------------------------------------------------------------- +# Tests +# --------------------------------------------------------------------------- + +@pytest.mark.asyncio +@pytest.mark.skipif(skip_database_tests, reason="skipping database tests") +@pytest.mark.skipif(not have_sync_redis, reason="skipping: sync redis not available for test setup") +async def test_get_metadata_valid_value(prefix): + expected_timestamp = 1234567890 + set_metadata(prefix, BigSegmentStoreMetadata(expected_timestamp)) + store = make_store(prefix) + try: + actual = await store.get_metadata() + assert actual is not None + assert actual.last_up_to_date == expected_timestamp + finally: + await store.stop() + + +@pytest.mark.asyncio +@pytest.mark.skipif(skip_database_tests, reason="skipping database tests") +@pytest.mark.skipif(not have_sync_redis, reason="skipping: sync redis not available for test setup") +async def test_get_metadata_no_value(prefix): + store = make_store(prefix) + try: + actual = await store.get_metadata() + assert actual is not None + assert actual.last_up_to_date is None + finally: + await store.stop() + + +@pytest.mark.asyncio +@pytest.mark.skipif(skip_database_tests, reason="skipping database tests") +@pytest.mark.skipif(not have_sync_redis, reason="skipping: sync redis not available for test setup") +async def test_get_membership_not_found(prefix): + store = make_store(prefix) + try: + membership = await store.get_membership(FAKE_USER_HASH) + assert membership is None + finally: + await store.stop() + + +@pytest.mark.asyncio +@pytest.mark.skipif(skip_database_tests, reason="skipping database tests") +@pytest.mark.skipif(not have_sync_redis, reason="skipping: sync redis not available for test setup") +async def test_get_membership_includes_only(prefix): + set_segments(prefix, FAKE_USER_HASH, ['key1', 'key2'], []) + store = make_store(prefix) + try: + membership = await store.get_membership(FAKE_USER_HASH) + assert membership == {'key1': True, 'key2': True} + finally: + await store.stop() + + +@pytest.mark.asyncio +@pytest.mark.skipif(skip_database_tests, reason="skipping database tests") +@pytest.mark.skipif(not have_sync_redis, reason="skipping: sync redis not available for test setup") +async def test_get_membership_excludes_only(prefix): + set_segments(prefix, FAKE_USER_HASH, [], ['key1', 'key2']) + store = make_store(prefix) + try: + membership = await store.get_membership(FAKE_USER_HASH) + assert membership == {'key1': False, 'key2': False} + finally: + await store.stop() + + +@pytest.mark.asyncio +@pytest.mark.skipif(skip_database_tests, reason="skipping database tests") +@pytest.mark.skipif(not have_sync_redis, reason="skipping: sync redis not available for test setup") +async def test_get_membership_includes_and_excludes(prefix): + set_segments(prefix, FAKE_USER_HASH, ['key1', 'key2'], ['key2', 'key3']) + store = make_store(prefix) + try: + membership = await store.get_membership(FAKE_USER_HASH) + assert membership == {'key1': True, 'key2': True, 'key3': False} + finally: + await store.stop() + + +@pytest.mark.asyncio +@pytest.mark.skipif(skip_database_tests, reason="skipping database tests") +@pytest.mark.skipif(not have_sync_redis, reason="skipping: sync redis not available for test setup") +async def test_stop_closes_client(prefix): + store = make_store(prefix) + # stop() should not raise + await store.stop() diff --git a/pyproject.toml b/pyproject.toml index b0888fe9..b62410d6 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -47,7 +47,7 @@ dev = [ "mock>=2.0.0", "pytest>=8.0.0,<10", "pytest-asyncio>=0.23", - "redis>=2.10.5,<9.0.0", + "redis>=4.2.0,<9.0.0", "boto3>=1.9.71,<2.0.0", "coverage>=4.4", "jsonpickle>1.4.1",