diff --git a/ldclient/migrations/__init__.py b/ldclient/migrations/__init__.py index 53a6cec0..2e1269b3 100644 --- a/ldclient/migrations/__init__.py +++ b/ldclient/migrations/__init__.py @@ -1,8 +1,15 @@ +# async_migrator is import-cheap (asyncio stdlib only, no aiohttp), so it is +# exported eagerly alongside the sync surface and keeps `import ldclient` cheap. +from .async_migrator import * from .migrator import * from .tracker import * from .types import * __all__ = [ + 'AsyncMigrationConfig', + 'AsyncMigrator', + 'AsyncMigratorBuilder', + 'AsyncMigratorFn', 'Migrator', 'MigratorBuilder', 'MigratorCompareFn', diff --git a/ldclient/migrations/async_migrator.py b/ldclient/migrations/async_migrator.py new file mode 100644 index 00000000..9853100b --- /dev/null +++ b/ldclient/migrations/async_migrator.py @@ -0,0 +1,397 @@ +from __future__ import annotations + +import asyncio +from abc import ABC, abstractmethod +from datetime import datetime +from random import Random +from typing import ( + TYPE_CHECKING, + Any, + Awaitable, + Callable, + Optional, + Tuple, + Union +) + +from ldclient.impl.sampler import Sampler +from ldclient.impl.util import Result +from ldclient.migrations.tracker import OpTracker +from ldclient.migrations.types import ( + ExecutionOrder, + MigrationConfig, + MigratorCompareFn, + Operation, + OperationResult, + Origin, + Stage, + WriteResult, + _MigratorBuilderBase +) + +if TYPE_CHECKING: + from ldclient import Context + from ldclient.async_client import AsyncLDClient + +__all__ = [ + 'AsyncMigrator', + 'AsyncMigratorBuilder', + 'AsyncMigratorImpl', + 'AsyncMigrationConfig', + 'AsyncExecutor', + 'AsyncMigratorFn', +] + +AsyncMigratorFn = Callable[[Optional[Any]], Awaitable[Any]] +""" +The async counterpart to :data:`ldclient.migrations.MigratorFn`. When an async +migration wishes to execute a read or write operation, it must delegate that +call to a consumer defined coroutine function. This function must accept an +optional payload value, and return a :class:`ldclient.Result`. +""" + + +class AsyncMigrator(ABC): + """ + An async migrator is the interface through which migration support is + executed for the async SDK. An async migrator is configured through the + :class:`AsyncMigratorBuilder`. + + .. 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. + """ + + @abstractmethod + async def read(self, key: str, context: Context, default_stage: Stage, payload: Optional[Any] = None) -> OperationResult: + """ + Uses the provided flag key and context to execute a migration-backed read operation. + + :param key: The migration flag key to use when determining the current stage + :param context: The context to use when evaluating the flag + :param default_stage: A default stage to fallback to if one cannot be determined + :param payload: An optional payload to be passed through to the appropriate read method + """ + + @abstractmethod + async def write(self, key: str, context: Context, default_stage: Stage, payload: Optional[Any] = None) -> WriteResult: + """ + Uses the provided flag key and context to execute a migration-backed write operation. + + :param key: The migration flag key to use when determining the current stage + :param context: The context to use when evaluating the flag + :param default_stage: A default stage to fallback to if one cannot be determined + :param payload: An optional payload to be passed through to the appropriate write method + """ + + +class AsyncMigratorImpl(AsyncMigrator): + """ + An implementation of the :class:`ldclient.migrations.AsyncMigrator` + interface, capable of supporting feature-flag backed technology migrations + for the async SDK. + """ + + def __init__( + self, + sampler: Sampler, + client: AsyncLDClient, + read_execution_order: ExecutionOrder, + read_config: AsyncMigrationConfig, + write_config: AsyncMigrationConfig, + measure_latency: bool, + measure_errors: bool, + ): + self._sampler = sampler + self._client = client + self._read_execution_order = read_execution_order + self._read_config = read_config + self._write_config = write_config + self._measure_latency = measure_latency + self._measure_errors = measure_errors + + async def read(self, key: str, context: Context, default_stage: Stage, payload: Optional[Any] = None) -> OperationResult: + stage, tracker = await self._client.migration_variation(key, context, default_stage) + tracker.operation(Operation.READ) + + old = AsyncExecutor(Origin.OLD, self._read_config.old, tracker, self._measure_latency, self._measure_errors, payload) + new = AsyncExecutor(Origin.NEW, self._read_config.new, tracker, self._measure_latency, self._measure_errors, payload) + + if stage == Stage.OFF: + result = await old.run() + elif stage == Stage.DUALWRITE: + result = await old.run() + elif stage == Stage.SHADOW: + result = await self.__read_both(old, new, tracker) + elif stage == Stage.LIVE: + result = await self.__read_both(new, old, tracker) + elif stage == Stage.RAMPDOWN: + result = await new.run() + else: + result = await new.run() + + # track_migration_op is synchronous on the async client; do not await it. + self._client.track_migration_op(tracker) + + return result + + async def write(self, key: str, context: Context, default_stage: Stage, payload: Optional[Any] = None) -> WriteResult: + stage, tracker = await self._client.migration_variation(key, context, default_stage) + tracker.operation(Operation.WRITE) + + old = AsyncExecutor(Origin.OLD, self._write_config.old, tracker, self._measure_latency, self._measure_errors, payload) + new = AsyncExecutor(Origin.NEW, self._write_config.new, tracker, self._measure_latency, self._measure_errors, payload) + + if stage == Stage.OFF: + result = await old.run() + write_result = WriteResult(result) + elif stage == Stage.DUALWRITE: + authoritative_result, nonauthoritative_result = await self.__write_both(old, new, tracker) + write_result = WriteResult(authoritative_result, nonauthoritative_result) + elif stage == Stage.SHADOW: + authoritative_result, nonauthoritative_result = await self.__write_both(old, new, tracker) + write_result = WriteResult(authoritative_result, nonauthoritative_result) + elif stage == Stage.LIVE: + authoritative_result, nonauthoritative_result = await self.__write_both(new, old, tracker) + write_result = WriteResult(authoritative_result, nonauthoritative_result) + elif stage == Stage.RAMPDOWN: + authoritative_result, nonauthoritative_result = await self.__write_both(new, old, tracker) + write_result = WriteResult(authoritative_result, nonauthoritative_result) + else: + result = await new.run() + write_result = WriteResult(result) + + # track_migration_op is synchronous on the async client; do not await it. + self._client.track_migration_op(tracker) + + return write_result + + async def __read_both(self, authoritative: AsyncExecutor, nonauthoritative: AsyncExecutor, tracker: OpTracker) -> OperationResult: + if self._read_execution_order == ExecutionOrder.PARALLEL: + authoritative_result, nonauthoritative_result = await asyncio.gather( + authoritative.run(), + nonauthoritative.run(), + ) + elif self._read_execution_order == ExecutionOrder.RANDOM and self._sampler.sample(2): + nonauthoritative_result = await nonauthoritative.run() + authoritative_result = await authoritative.run() + else: + authoritative_result = await authoritative.run() + nonauthoritative_result = await nonauthoritative.run() + + if self._read_config.comparison is None: + return authoritative_result + + compare = self._read_config.comparison + if authoritative_result.is_success() and nonauthoritative_result.is_success(): + tracker.consistent(lambda: compare(authoritative_result.value, nonauthoritative_result.value)) + + return authoritative_result + + async def __write_both(self, authoritative: AsyncExecutor, nonauthoritative: AsyncExecutor, tracker: OpTracker) -> Tuple[OperationResult, Optional[OperationResult]]: + authoritative_result = await authoritative.run() + tracker.invoked(authoritative.origin) + + if not authoritative_result.is_success(): + return authoritative_result, None + + nonauthoritative_result = await nonauthoritative.run() + tracker.invoked(nonauthoritative.origin) + + return authoritative_result, nonauthoritative_result + + +class AsyncMigrationConfig: + """ + The async counterpart to :class:`ldclient.migrations.MigrationConfig`. It + stores references to coroutine functions which execute customer defined + read or write operations on old or new origins of information. For read + operations, an optional (synchronous) comparison function can also be + defined. + + .. 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. + """ + + def __init__(self, old: AsyncMigratorFn, new: AsyncMigratorFn, comparison: Optional[MigratorCompareFn] = None): + self.__old = old + self.__new = new + self.__comparison = comparison + + @property + def old(self) -> AsyncMigratorFn: + """ + Coroutine function which receives a nullable payload parameter and + returns an awaitable resolving to an :class:`ldclient.Result`. + + This function call should affect the old migration origin when called. + """ + return self.__old + + @property + def new(self) -> AsyncMigratorFn: + """ + Coroutine function which receives a nullable payload parameter and + returns an awaitable resolving to an :class:`ldclient.Result`. + + This function call should affect the new migration origin when called. + """ + return self.__new + + @property + def comparison(self) -> Optional[MigratorCompareFn]: + """ + Optional (synchronous) callable which receives two objects of any kind + and returns a boolean representing equality. + + The result of this comparison can be sent upstream to LaunchDarkly to + enhance migration observability. + """ + return self.__comparison + + +class AsyncMigratorBuilder(_MigratorBuilderBase): + """ + The async migration builder is used to configure and construct an instance + of an :class:`AsyncMigrator`. This migrator can be used to perform + LaunchDarkly assisted technology migrations through the use of + migration-based feature flags. + + .. 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. + """ + + def __init__(self, client: AsyncLDClient): + # Single _ to prevent mangling; useful for testing + self._client = client + + # Default settings as required by the spec + self._read_execution_order = ExecutionOrder.PARALLEL + self._measure_latency = True + self._measure_errors = True + + self.__read_config: Optional[AsyncMigrationConfig] = None + self.__write_config: Optional[AsyncMigrationConfig] = None + + def read(self, old: AsyncMigratorFn, new: AsyncMigratorFn, comparison: Optional[MigratorCompareFn] = None) -> 'AsyncMigratorBuilder': + """ + Read can be used to configure the migration-read behavior of the + resulting :class:`AsyncMigrator` instance. + + Users are required to provide two different read coroutine functions -- + one to read from the old migration origin, and one to read from the new + origin. Additionally, customers can opt-in to consistency tracking by + providing a comparison function. + + Depending on the migration stage, one or both of these read methods may + be called. + + The read methods should accept a single nullable parameter. This + parameter is a payload passed through the :func:`AsyncMigrator.read` + method. This method should return a :class:`ldclient.Result` instance. + + The consistency method should accept 2 parameters of any type. These + parameters are the results of executing the read operation against the + old and new origins. If both operations were successful, the + consistency method will be invoked. This method should return true if + the two parameters are equal, or false otherwise. The comparison + function is synchronous. + + :param old: The coroutine function to execute when reading from the old origin + :param new: The coroutine function to execute when reading from the new origin + :param comparison: An optional function to use for comparing the results from two origins + """ + self.__read_config = AsyncMigrationConfig(old, new, comparison) + return self + + def write(self, old: AsyncMigratorFn, new: AsyncMigratorFn) -> 'AsyncMigratorBuilder': + """ + Write can be used to configure the migration-write behavior of the + resulting :class:`AsyncMigrator` instance. + + Users are required to provide two different write coroutine functions -- + one to write to the old migration origin, and one to write to the new + origin. + + Depending on the migration stage, one or both of these write methods + may be called. + + The write methods should accept a single nullable parameter. This + parameter is a payload passed through the :func:`AsyncMigrator.write` + method. This method should return a :class:`ldclient.Result` instance. + + :param old: The coroutine function to execute when writing to the old origin + :param new: The coroutine function to execute when writing to the new origin + """ + self.__write_config = AsyncMigrationConfig(old, new) + return self + + def build(self) -> Union[AsyncMigrator, str]: + """ + Build constructs an :class:`AsyncMigrator` instance to support + migration-based reads and writes. A string describing any failure + conditions will be returned if the build fails. + """ + if self.__read_config is None: + return "read configuration not provided" + + if self.__write_config is None: + return "write configuration not provided" + + return AsyncMigratorImpl( + Sampler(Random()), + self._client, + self._read_execution_order, + self.__read_config, + self.__write_config, + self._measure_latency, + self._measure_errors, + ) + + +class AsyncExecutor: + """ + Utility class for executing async migration operations while also tracking + our built-in migration measurements. + """ + + def __init__(self, origin: Origin, fn: AsyncMigratorFn, tracker: OpTracker, measure_latency: bool, measure_errors: bool, payload: Any): + self.__origin = origin + self.__fn = fn + self.__tracker = tracker + self.__measure_latency = measure_latency + self.__measure_errors = measure_errors + self.__payload = payload + + @property + def origin(self) -> Origin: + return self.__origin + + async def run(self) -> OperationResult: + """ + Execute the configured operation and track any available measurements. + """ + start = datetime.now() + + try: + result = await self.__fn(self.__payload) + except Exception as e: + result = Result.fail(f"'{self.__origin.value} operation raised an exception", e) + + # Record required tracker measurements + if self.__measure_latency: + self.__tracker.latency(self.__origin, datetime.now() - start) + + if self.__measure_errors and not result.is_success(): + self.__tracker.error(self.__origin) + + self.__tracker.invoked(self.__origin) + + return OperationResult(self.__origin, result) diff --git a/ldclient/migrations/migrator.py b/ldclient/migrations/migrator.py index fde023f0..1b23e464 100644 --- a/ldclient/migrations/migrator.py +++ b/ldclient/migrations/migrator.py @@ -18,7 +18,8 @@ OperationResult, Origin, Stage, - WriteResult + WriteResult, + _MigratorBuilderBase ) if TYPE_CHECKING: @@ -168,7 +169,7 @@ def __write_both(self, authoritative: Executor, nonauthoritative: Executor, trac return authoritative_result, nonauthoritative_result -class MigratorBuilder: +class MigratorBuilder(_MigratorBuilderBase): """ The migration builder is used to configure and construct an instance of a :class:`Migrator`. This migrator can be used to perform LaunchDarkly @@ -181,42 +182,13 @@ def __init__(self, client: LDClient): self._client = client # Default settings as required by the spec - self.__read_execution_order = ExecutionOrder.PARALLEL - self.__measure_latency = True - self.__measure_errors = True + self._read_execution_order = ExecutionOrder.PARALLEL + self._measure_latency = True + self._measure_errors = True self.__read_config: Optional[MigrationConfig] = None self.__write_config: Optional[MigrationConfig] = None - def read_execution_order(self, order: ExecutionOrder) -> 'MigratorBuilder': - """ - The read execution order influences the parallelism and execution order - for read operations involving multiple origins. - """ - if order not in ExecutionOrder: - return self - - self.__read_execution_order = order - return self - - def track_latency(self, enabled: bool) -> 'MigratorBuilder': - """ - Enable or disable latency tracking for migration operations. This - latency information can be sent upstream to LaunchDarkly to enhance - migration visibility. - """ - self.__measure_latency = enabled - return self - - def track_errors(self, enabled: bool) -> 'MigratorBuilder': - """ - Enable or disable error tracking for migration operations. This error - information can be sent upstream to LaunchDarkly to enhance migration - visibility. - """ - self.__measure_errors = enabled - return self - def read(self, old: MigratorFn, new: MigratorFn, comparison: Optional[MigratorCompareFn] = None) -> 'MigratorBuilder': """ Read can be used to configure the migration-read behavior of the @@ -283,11 +255,11 @@ def build(self) -> Union[Migrator, str]: return MigratorImpl( Sampler(Random()), self._client, - self.__read_execution_order, + self._read_execution_order, self.__read_config, self.__write_config, - self.__measure_latency, - self.__measure_errors, + self._measure_latency, + self._measure_errors, ) diff --git a/ldclient/migrations/types.py b/ldclient/migrations/types.py index 295731ea..2fcffbc6 100644 --- a/ldclient/migrations/types.py +++ b/ldclient/migrations/types.py @@ -1,5 +1,5 @@ from enum import Enum -from typing import Any, Callable, Optional +from typing import Any, Callable, Optional, TypeVar from ldclient.impl.util import Result @@ -242,3 +242,48 @@ def comparison(self) -> Optional[MigratorCompareFn]: enhance migration observability. """ return self.__comparison + + +_MigratorBuilderT = TypeVar('_MigratorBuilderT', bound='_MigratorBuilderBase') + + +class _MigratorBuilderBase: + """ + Shared setter implementation for :class:`MigratorBuilder` and its async + counterpart. It holds the fluent configuration methods common to both + builders. The ``read``, ``write``, and ``build`` methods differ between the + sync and async builders and are defined on each subclass. + """ + + _read_execution_order: ExecutionOrder + _measure_latency: bool + _measure_errors: bool + + def read_execution_order(self: _MigratorBuilderT, order: ExecutionOrder) -> _MigratorBuilderT: + """ + The read execution order influences the parallelism and execution order + for read operations involving multiple origins. + """ + if order not in ExecutionOrder: + return self + + self._read_execution_order = order + return self + + def track_latency(self: _MigratorBuilderT, enabled: bool) -> _MigratorBuilderT: + """ + Enable or disable latency tracking for migration operations. This + latency information can be sent upstream to LaunchDarkly to enhance + migration visibility. + """ + self._measure_latency = enabled + return self + + def track_errors(self: _MigratorBuilderT, enabled: bool) -> _MigratorBuilderT: + """ + Enable or disable error tracking for migration operations. This error + information can be sent upstream to LaunchDarkly to enhance migration + visibility. + """ + self._measure_errors = enabled + return self diff --git a/ldclient/testing/migrations/test_async_migrator.py b/ldclient/testing/migrations/test_async_migrator.py new file mode 100644 index 00000000..db6e1f73 --- /dev/null +++ b/ldclient/testing/migrations/test_async_migrator.py @@ -0,0 +1,589 @@ +import asyncio +from datetime import datetime, timedelta +from typing import List + +import pytest + +from ldclient import Result +from ldclient.context import Context +from ldclient.evaluation import EvaluationDetail +from ldclient.impl.events.types import EventInputEvaluation +from ldclient.impl.model import FeatureFlag +from ldclient.impl.util import timedelta_millis +from ldclient.migrations import ( + AsyncMigrator, + AsyncMigratorBuilder, + AsyncMigratorFn +) +from ldclient.migrations.tracker import MigrationOpEvent, OpTracker +from ldclient.migrations.types import ExecutionOrder, Origin, Stage +from ldclient.testing.builders import FlagBuilder + +user = Context.from_dict({u'key': u'xyz', u'kind': u'user', u'bizzle': u'def'}) + + +class FakeEventProcessor: + """Records events the way the real (async/sync) event processors expose them + so tests can assert on the emitted MigrationOpEvent, mirroring the sync + migrator tests' use of ``client._event_processor._events``.""" + + def __init__(self): + self._events: List = [] + + def send_event(self, event): + self._events.append(event) + + +class FakeAsyncClient: + """A minimal stand-in for AsyncLDClient that exposes the exact surface the + AsyncMigrator depends on: an async ``migration_variation`` returning + ``(Stage, OpTracker)`` and a synchronous ``track_migration_op``. + + The flag key passed to ``migration_variation`` is interpreted as a Stage + value (matching the sync migrator tests), and a real OpTracker is built from + a real FeatureFlag so the consistency/error/latency/event-build semantics + are genuinely exercised. + """ + + def __init__(self): + self._event_processor = FakeEventProcessor() + self._flags = {} + for stage in Stage: + flag = FlagBuilder(stage.value).on(True).variations(stage.value).fallthrough_variation(0).build() + self._flags[stage.value] = flag + + async def migration_variation(self, key: str, context: Context, default_stage: Stage): + # Yield once to confirm callers truly await this coroutine. + await asyncio.sleep(0) + flag: FeatureFlag = self._flags[key] + stage = Stage(key) + detail = EvaluationDetail(stage.value, 0, {'kind': 'FALLTHROUGH'}) + tracker = OpTracker(key, flag, context, detail, default_stage) + return stage, tracker + + def track_migration_op(self, tracker: OpTracker): + # Synchronous on the real async client; must NOT be awaited. + event = tracker.build() + if isinstance(event, str): + raise AssertionError("tracker.build() failed: %s" % event) + # Emulate the EventInputEvaluation that migration_variation would have + # queued, so index [1] is the MigrationOpEvent (as in the sync tests). + self._event_processor.send_event(_FakeEvalEvent()) + self._event_processor.send_event(event) + + +class _FakeEvalEvent(EventInputEvaluation): + def __init__(self): + pass + + +async def async_success(payload) -> Result: + return Result.success(True) + + +def raises_exception(msg) -> AsyncMigratorFn: + async def inner(payload): + raise Exception(msg) + + return inner + + +@pytest.fixture +def builder() -> AsyncMigratorBuilder: + client = FakeAsyncClient() + builder = AsyncMigratorBuilder(client) # type: ignore[arg-type] + builder.track_latency(False) + builder.track_errors(False) + + builder.read(async_success, async_success, None) + builder.write(async_success, async_success) + + return builder + + +class TestBuilder: + def test_can_build_successfully(self): + client = FakeAsyncClient() + builder = AsyncMigratorBuilder(client) # type: ignore[arg-type] + builder.read(async_success, async_success, None) + builder.write(async_success, async_success) + migrator = builder.build() + assert isinstance(migrator, AsyncMigrator) + + @pytest.mark.parametrize( + "order", + [ + pytest.param(ExecutionOrder.SERIAL, id="serial"), + pytest.param(ExecutionOrder.RANDOM, id="random"), + pytest.param(ExecutionOrder.PARALLEL, id="parallel"), + ], + ) + def test_can_modify_execution_order(self, order): + client = FakeAsyncClient() + builder = AsyncMigratorBuilder(client) # type: ignore[arg-type] + builder.read(async_success, async_success, None) + builder.write(async_success, async_success) + builder.read_execution_order(order) + migrator = builder.build() + assert isinstance(migrator, AsyncMigrator) + + def test_build_fails_without_read(self): + client = FakeAsyncClient() + builder = AsyncMigratorBuilder(client) # type: ignore[arg-type] + builder.write(async_success, async_success) + migrator = builder.build() + assert isinstance(migrator, str) + assert migrator == "read configuration not provided" + + def test_build_fails_without_write(self): + client = FakeAsyncClient() + builder = AsyncMigratorBuilder(client) # type: ignore[arg-type] + builder.read(async_success, async_success) + migrator = builder.build() + assert isinstance(migrator, str) + assert migrator == "write configuration not provided" + + +@pytest.mark.asyncio +class TestPassingPayloadThrough: + @pytest.mark.parametrize( + "stage,count", + [ + pytest.param(Stage.OFF, 1, id="off"), + pytest.param(Stage.DUALWRITE, 1, id="dualwrite"), + pytest.param(Stage.SHADOW, 2, id="shadow"), + pytest.param(Stage.LIVE, 2, id="live"), + pytest.param(Stage.RAMPDOWN, 1, id="rampdown"), + pytest.param(Stage.COMPLETE, 1, id="complete"), + ], + ) + async def test_passes_through_read(self, builder: AsyncMigratorBuilder, stage: Stage, count: int): + payloads = [] + + async def capture_payloads(payload): + payloads.append(payload) + return Result.success(None) + + builder.read(capture_payloads, capture_payloads) + migrator = builder.build() + assert isinstance(migrator, AsyncMigrator) + + result = await migrator.read(stage.value, user, Stage.LIVE, "payload") + + assert result.is_success() + assert len(payloads) == count + assert all("payload" == p for p in payloads) + + @pytest.mark.parametrize( + "stage,count", + [ + pytest.param(Stage.OFF, 1, id="off"), + pytest.param(Stage.DUALWRITE, 2, id="dualwrite"), + pytest.param(Stage.SHADOW, 2, id="shadow"), + pytest.param(Stage.LIVE, 2, id="live"), + pytest.param(Stage.RAMPDOWN, 2, id="rampdown"), + pytest.param(Stage.COMPLETE, 1, id="complete"), + ], + ) + async def test_passes_through_write(self, builder: AsyncMigratorBuilder, stage: Stage, count: int): + payloads = [] + + async def capture_payloads(payload): + payloads.append(payload) + return Result.success(None) + + builder.write(capture_payloads, capture_payloads) + migrator = builder.build() + assert isinstance(migrator, AsyncMigrator) + + result = await migrator.write(stage.value, user, Stage.LIVE, "payload") + + assert result.authoritative.is_success() + if result.nonauthoritative is not None: + assert result.nonauthoritative.is_success() + + assert len(payloads) == count + assert all("payload" == p for p in payloads) + + +@pytest.mark.asyncio +class TestTrackingInvoked: + @pytest.mark.parametrize( + "stage,origins", + [ + pytest.param(Stage.OFF, [Origin.OLD], id="off"), + pytest.param(Stage.DUALWRITE, [Origin.OLD], id="dualwrite"), + pytest.param(Stage.SHADOW, [Origin.OLD, Origin.NEW], id="shadow"), + pytest.param(Stage.LIVE, [Origin.OLD, Origin.NEW], id="live"), + pytest.param(Stage.RAMPDOWN, [Origin.NEW], id="rampdown"), + pytest.param(Stage.COMPLETE, [Origin.NEW], id="complete"), + ], + ) + async def test_reads(self, builder: AsyncMigratorBuilder, stage: Stage, origins: List[Origin]): + migrator = builder.build() + assert isinstance(migrator, AsyncMigrator) + + result = await migrator.read(stage.value, user, Stage.LIVE) + + assert result.is_success() + event = builder._client._event_processor._events[1] # type: ignore + assert isinstance(event, MigrationOpEvent) + assert len(origins) == len(event.invoked) + assert all(o in event.invoked for o in origins) + + @pytest.mark.parametrize( + "stage,origins", + [ + pytest.param(Stage.OFF, [Origin.OLD], id="off"), + pytest.param(Stage.DUALWRITE, [Origin.OLD, Origin.NEW], id="dualwrite"), + pytest.param(Stage.SHADOW, [Origin.OLD, Origin.NEW], id="shadow"), + pytest.param(Stage.LIVE, [Origin.OLD, Origin.NEW], id="live"), + pytest.param(Stage.RAMPDOWN, [Origin.OLD, Origin.NEW], id="rampdown"), + pytest.param(Stage.COMPLETE, [Origin.NEW], id="complete"), + ], + ) + async def test_writes(self, builder: AsyncMigratorBuilder, stage: Stage, origins: List[Origin]): + migrator = builder.build() + assert isinstance(migrator, AsyncMigrator) + + result = await migrator.write(stage.value, user, Stage.LIVE) + + assert result.authoritative.is_success() + event = builder._client._event_processor._events[1] # type: ignore + assert isinstance(event, MigrationOpEvent) + assert len(origins) == len(event.invoked) + assert all(o in event.invoked for o in origins) + + +@pytest.mark.asyncio +class TestTrackingLatency: + @pytest.mark.parametrize( + "stage,origins", + [ + pytest.param(Stage.OFF, [Origin.OLD], id="off"), + pytest.param(Stage.DUALWRITE, [Origin.OLD], id="dualwrite"), + pytest.param(Stage.SHADOW, [Origin.OLD, Origin.NEW], id="shadow"), + pytest.param(Stage.LIVE, [Origin.OLD, Origin.NEW], id="live"), + pytest.param(Stage.RAMPDOWN, [Origin.NEW], id="rampdown"), + pytest.param(Stage.COMPLETE, [Origin.NEW], id="complete"), + ], + ) + async def test_reads(self, builder: AsyncMigratorBuilder, stage: Stage, origins: List[Origin]): + async def delay(payload): + await asyncio.sleep(0.1) + return Result.success("success") + + builder.track_latency(True) + builder.read(delay, delay) + migrator = builder.build() + assert isinstance(migrator, AsyncMigrator) + + result = await migrator.read(stage.value, user, Stage.LIVE) + + assert result.is_success() + event = builder._client._event_processor._events[1] # type: ignore + assert isinstance(event, MigrationOpEvent) + assert len(origins) == len(event.latencies) + for o in origins: + assert o in event.latencies + assert event.latencies[o] >= timedelta(milliseconds=100) + + @pytest.mark.parametrize( + "stage,origins", + [ + pytest.param(Stage.OFF, [Origin.OLD], id="off"), + pytest.param(Stage.DUALWRITE, [Origin.OLD, Origin.NEW], id="dualwrite"), + pytest.param(Stage.SHADOW, [Origin.OLD, Origin.NEW], id="shadow"), + pytest.param(Stage.LIVE, [Origin.OLD, Origin.NEW], id="live"), + pytest.param(Stage.RAMPDOWN, [Origin.OLD, Origin.NEW], id="rampdown"), + pytest.param(Stage.COMPLETE, [Origin.NEW], id="complete"), + ], + ) + async def test_writes(self, builder: AsyncMigratorBuilder, stage: Stage, origins: List[Origin]): + async def delay(payload): + await asyncio.sleep(0.1) + return Result.success("success") + + builder.track_latency(True) + builder.write(delay, delay) + migrator = builder.build() + assert isinstance(migrator, AsyncMigrator) + + result = await migrator.write(stage.value, user, Stage.LIVE) + + assert result.authoritative.is_success() + event = builder._client._event_processor._events[1] # type: ignore + assert isinstance(event, MigrationOpEvent) + assert len(origins) == len(event.latencies) + for o in origins: + assert o in event.latencies + assert event.latencies[o] >= timedelta(milliseconds=100) + + +@pytest.mark.asyncio +class TestTrackingErrors: + @pytest.mark.parametrize( + "stage,origins", + [ + pytest.param(Stage.OFF, [Origin.OLD], id="off"), + pytest.param(Stage.DUALWRITE, [Origin.OLD], id="dualwrite"), + pytest.param(Stage.SHADOW, [Origin.OLD, Origin.NEW], id="shadow"), + pytest.param(Stage.LIVE, [Origin.OLD, Origin.NEW], id="live"), + pytest.param(Stage.RAMPDOWN, [Origin.NEW], id="rampdown"), + pytest.param(Stage.COMPLETE, [Origin.NEW], id="complete"), + ], + ) + async def test_reads(self, builder: AsyncMigratorBuilder, stage: Stage, origins: List[Origin]): + async def fail(_): + return Result.fail("fail") + + builder.track_errors(True) + builder.read(fail, fail) + migrator = builder.build() + assert isinstance(migrator, AsyncMigrator) + + result = await migrator.read(stage.value, user, Stage.LIVE) + + assert not result.is_success() + event = builder._client._event_processor._events[1] # type: ignore + assert isinstance(event, MigrationOpEvent) + assert len(origins) == len(event.errors) + assert all(o in event.errors for o in origins) + + @pytest.mark.parametrize( + "stage,origin", + [ + pytest.param(Stage.OFF, Origin.OLD, id="off"), + pytest.param(Stage.DUALWRITE, Origin.OLD, id="dualwrite"), + pytest.param(Stage.SHADOW, Origin.OLD, id="shadow"), + pytest.param(Stage.LIVE, Origin.NEW, id="live"), + pytest.param(Stage.RAMPDOWN, Origin.NEW, id="rampdown"), + pytest.param(Stage.COMPLETE, Origin.NEW, id="complete"), + ], + ) + async def test_authoritative_writes(self, builder: AsyncMigratorBuilder, stage: Stage, origin: Origin): + async def fail(_): + return Result.fail("fail") + + builder.track_errors(True) + builder.write(fail, fail) + migrator = builder.build() + assert isinstance(migrator, AsyncMigrator) + + result = await migrator.write(stage.value, user, Stage.LIVE) + + assert not result.authoritative.is_success() + assert result.nonauthoritative is None + event = builder._client._event_processor._events[1] # type: ignore + assert isinstance(event, MigrationOpEvent) + assert 1 == len(event.errors) + assert origin in event.errors + + @pytest.mark.parametrize( + "stage,fail_old,fail_new,origin", + [ + pytest.param(Stage.DUALWRITE, False, True, Origin.NEW, id="dualwrite"), + pytest.param(Stage.SHADOW, False, True, Origin.NEW, id="shadow"), + pytest.param(Stage.LIVE, True, False, Origin.OLD, id="live"), + pytest.param(Stage.RAMPDOWN, True, False, Origin.OLD, id="rampdown"), + ], + ) + async def test_nonauthoritative_writes(self, builder: AsyncMigratorBuilder, stage: Stage, fail_old: bool, fail_new: bool, origin: Origin): + async def success(_): + return Result.success(None) + + async def fail(_): + return Result.fail("fail") + + builder.track_errors(True) + builder.write(fail if fail_old else success, fail if fail_new else success) + migrator = builder.build() + assert isinstance(migrator, AsyncMigrator) + + result = await migrator.write(stage.value, user, Stage.LIVE) + + assert result.authoritative.is_success() + assert result.nonauthoritative is not None + assert not result.nonauthoritative.is_success() + event = builder._client._event_processor._events[1] # type: ignore + assert isinstance(event, MigrationOpEvent) + assert 1 == len(event.errors) + assert origin in event.errors + + +@pytest.mark.asyncio +class TestTrackingConsistency: + @pytest.mark.parametrize( + "stage", + [ + pytest.param(Stage.OFF, id="off"), + pytest.param(Stage.DUALWRITE, id="dualwrite"), + pytest.param(Stage.RAMPDOWN, id="rampdown"), + pytest.param(Stage.COMPLETE, id="complete"), + ], + ) + async def test_consistency_is_not_run_in_most_stages(self, builder: AsyncMigratorBuilder, stage: Stage): + async def value(_): + return Result.success("value") + + builder.read(value, value, lambda lhs, rhs: lhs == rhs) + migrator = builder.build() + assert isinstance(migrator, AsyncMigrator) + + result = await migrator.read(stage.value, user, Stage.LIVE) + assert result.is_success() + event = builder._client._event_processor._events[1] # type: ignore + assert isinstance(event, MigrationOpEvent) + assert event.consistent is None + + @pytest.mark.parametrize( + "stage,old,new,expected", + [ + pytest.param(Stage.SHADOW, "value", "value", True, id="shadow matches"), + pytest.param(Stage.LIVE, "value", "value", True, id="live matches"), + pytest.param(Stage.SHADOW, "old", "new", False, id="shadow does not match"), + pytest.param(Stage.LIVE, "old", "new", False, id="live does not match"), + ], + ) + async def test_consistency_is_tracked_correctly(self, builder: AsyncMigratorBuilder, stage: Stage, old: str, new: str, expected: bool): + async def old_fn(_): + return Result.success(old) + + async def new_fn(_): + return Result.success(new) + + builder.read(old_fn, new_fn, lambda lhs, rhs: lhs == rhs) + migrator = builder.build() + assert isinstance(migrator, AsyncMigrator) + + result = await migrator.read(stage.value, user, Stage.LIVE) + assert result.is_success() + event = builder._client._event_processor._events[1] # type: ignore + assert isinstance(event, MigrationOpEvent) + assert event.consistent is expected + + @pytest.mark.parametrize( + "stage,old,new", + [ + pytest.param(Stage.SHADOW, "value", "value", id="shadow"), + pytest.param(Stage.LIVE, "value", "value", id="live"), + ], + ) + async def test_consistency_handles_exceptions(self, builder: AsyncMigratorBuilder, stage: Stage, old: str, new: str): + def raise_exception(lhs, rhs): + raise Exception("error") + + async def old_fn(_): + return Result.success(old) + + async def new_fn(_): + return Result.success(new) + + builder.read(old_fn, new_fn, raise_exception) + migrator = builder.build() + assert isinstance(migrator, AsyncMigrator) + + result = await migrator.read(stage.value, user, Stage.LIVE) + assert result.is_success() + event = builder._client._event_processor._events[1] # type: ignore + assert isinstance(event, MigrationOpEvent) + assert event.consistent is None + + +@pytest.mark.asyncio +class TestHandlesExceptionsInMigratorFn: + @pytest.mark.parametrize( + "stage,expected_msg", + [ + pytest.param(Stage.OFF, "old read", id="off"), + pytest.param(Stage.DUALWRITE, "old read", id="dualwrite"), + pytest.param(Stage.SHADOW, "old read", id="shadow"), + pytest.param(Stage.LIVE, "new read", id="live"), + pytest.param(Stage.RAMPDOWN, "new read", id="rampdown"), + pytest.param(Stage.COMPLETE, "new read", id="complete"), + ], + ) + async def test_reads(self, builder: AsyncMigratorBuilder, stage: Stage, expected_msg: str): + builder.read(raises_exception("old read"), raises_exception("new read")) + migrator = builder.build() + assert isinstance(migrator, AsyncMigrator) + + result = await migrator.read(stage.value, user, Stage.LIVE) + + assert result.is_success() is False + assert str(result.exception) == expected_msg + + @pytest.mark.parametrize( + "stage,expected_msg", + [ + pytest.param(Stage.OFF, "old write", id="off"), + pytest.param(Stage.DUALWRITE, "old write", id="dualwrite"), + pytest.param(Stage.SHADOW, "old write", id="shadow"), + pytest.param(Stage.LIVE, "new write", id="live"), + pytest.param(Stage.RAMPDOWN, "new write", id="rampdown"), + pytest.param(Stage.COMPLETE, "new write", id="complete"), + ], + ) + async def test_exception_in_authoritative_write(self, builder: AsyncMigratorBuilder, stage: Stage, expected_msg: str): + builder.write(raises_exception("old write"), raises_exception("new write")) + migrator = builder.build() + assert isinstance(migrator, AsyncMigrator) + + result = await migrator.write(stage.value, user, Stage.LIVE) + + assert result.authoritative.is_success() is False + assert str(result.authoritative.exception) == expected_msg + assert result.nonauthoritative is None + + @pytest.mark.parametrize( + "stage,expected_msg,fail_old", + [ + pytest.param(Stage.DUALWRITE, "new write", False, id="dualwrite"), + pytest.param(Stage.SHADOW, "new write", False, id="shadow"), + pytest.param(Stage.LIVE, "old write", True, id="live"), + pytest.param(Stage.RAMPDOWN, "old write", True, id="rampdown"), + ], + ) + async def test_exception_in_nonauthoritative_write(self, builder: AsyncMigratorBuilder, stage: Stage, expected_msg: str, fail_old: bool): + old_fn = raises_exception("old write") if fail_old else async_success + new_fn = async_success if fail_old else raises_exception("new write") + + builder.write(old_fn, new_fn) + migrator = builder.build() + assert isinstance(migrator, AsyncMigrator) + + result = await migrator.write(stage.value, user, Stage.LIVE) + + assert result.authoritative.is_success() + assert result.nonauthoritative is not None + assert not result.nonauthoritative.is_success() + assert str(result.nonauthoritative.exception) == expected_msg + + +@pytest.mark.asyncio +class TestSupportsExecutionOrder: + @pytest.mark.parametrize( + "order,min_time", + [ + pytest.param(ExecutionOrder.PARALLEL, 300, id="parallel"), + pytest.param(ExecutionOrder.SERIAL, 600, id="serial"), + pytest.param(ExecutionOrder.RANDOM, 600, id="random"), + ], + ) + async def test_parallel(self, builder: AsyncMigratorBuilder, order: ExecutionOrder, min_time: int): + async def delay(payload): + await asyncio.sleep(0.3) + return Result.success("success") + + builder.read_execution_order(order) + builder.read(delay, delay) + migrator = builder.build() + assert isinstance(migrator, AsyncMigrator) + + start = datetime.now() + result = await migrator.read('live', user, Stage.LIVE) + delta = datetime.now() - start + ms = timedelta_millis(delta) + + assert result.is_success() + assert ms >= min_time