From 089c4b81a850a47b7197842043cc0e142d02ed9d Mon Sep 17 00:00:00 2001 From: Thomas Gaigher Date: Thu, 9 Jul 2026 07:43:13 +0000 Subject: [PATCH] feat(testing): add skip-time virtual clock - Add Clock protocol with RealClock and SkipClock implementations - SkipClock returns real time plus the accumulated skipped duration, so virtual time advances with real time between jumps and every read stays monotonic and distinct - One clock per runner, shared by the Executor, CheckpointProcessor, and all timestamp stamping, so history events form a single totally ordered clock domain - Drive wait and step-retry timers off the clock so modeled delays complete instantly under skip while history keeps modeled durations - Stamp operation, invocation, and execution start/end timestamps from the clock; reconstruct operations from event timestamps - Ban datetime.now via ruff TID251 outside clock.py so all time reads go through the clock - Default DurableFunctionTestRunner to skip_time=True; pass skip_time=False for real wall-clock timing - Remove the DURABLE_EXECUTION_TIME_SCALE env-var scaling --- .../test/conftest.py | 5 +- .../test/plugin/test_plugin.py | 4 +- .../pyproject.toml | 6 +- .../checkpoint/core.py | 8 ++ .../checkpoint/processor.py | 9 ++ .../checkpoint/processors/base.py | 41 ++++--- .../checkpoint/processors/callback.py | 7 +- .../checkpoint/processors/context.py | 6 + .../checkpoint/processors/execution.py | 3 + .../checkpoint/processors/step.py | 12 +- .../checkpoint/processors/wait.py | 15 +-- .../checkpoint/transformer.py | 6 +- .../cli.py | 7 ++ .../client.py | 3 +- .../clock.py | 69 ++++++++++++ .../execution.py | 50 +++++---- .../executor.py | 47 ++++---- .../model.py | 12 +- .../runner.py | 22 +++- .../tests/checkpoint/processors/base_test.py | 29 +++-- .../checkpoint/processors/callback_test.py | 36 +++++- .../checkpoint/processors/context_test.py | 40 ++++--- .../processors/execution_processor_test.py | 25 +++-- .../tests/checkpoint/processors/step_test.py | 46 +++++--- .../tests/checkpoint/processors/wait_test.py | 28 ++--- .../tests/checkpoint/transformer_test.py | 17 ++- .../tests/clock_test.py | 88 +++++++++++++++ .../tests/execution_test.py | 12 +- .../tests/executor_test.py | 20 ++-- .../tests/skip_time_test.py | 105 ++++++++++++++++++ 30 files changed, 602 insertions(+), 176 deletions(-) create mode 100644 packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/clock.py create mode 100644 packages/aws-durable-execution-sdk-python-testing/tests/clock_test.py create mode 100644 packages/aws-durable-execution-sdk-python-testing/tests/skip_time_test.py diff --git a/packages/aws-durable-execution-sdk-python-examples/test/conftest.py b/packages/aws-durable-execution-sdk-python-examples/test/conftest.py index 31d32b6f..84eda88c 100644 --- a/packages/aws-durable-execution-sdk-python-examples/test/conftest.py +++ b/packages/aws-durable-execution-sdk-python-examples/test/conftest.py @@ -230,8 +230,11 @@ def test_hello_world(durable_runner): else: if not handler: pytest.fail("handler is required for local mode tests") + skip_time = marker.kwargs.get("skip_time", True) # Create local runner (needs cleanup via context manager) - runner = DurableFunctionTestRunner(handler=handler, execution_timeout=60) + runner = DurableFunctionTestRunner( + handler=handler, execution_timeout=60, skip_time=skip_time + ) # Wrap in adapter and use context manager for proper cleanup with TestRunnerAdapter(runner, runner_mode) as adapter: diff --git a/packages/aws-durable-execution-sdk-python-examples/test/plugin/test_plugin.py b/packages/aws-durable-execution-sdk-python-examples/test/plugin/test_plugin.py index 28d041b0..888f661d 100644 --- a/packages/aws-durable-execution-sdk-python-examples/test/plugin/test_plugin.py +++ b/packages/aws-durable-execution-sdk-python-examples/test/plugin/test_plugin.py @@ -34,10 +34,8 @@ def test_plugin(durable_runner): lambda_function_name="Plugin Wait", ) def test_plugin_on_operation_end_called_for_wait_completed_during_suspend( - durable_runner, monkeypatch + durable_runner, ): - monkeypatch.setenv("DURABLE_EXECUTION_TIME_SCALE", "0.01") - with durable_runner: result = durable_runner.run(input=None, timeout=30) diff --git a/packages/aws-durable-execution-sdk-python-testing/pyproject.toml b/packages/aws-durable-execution-sdk-python-testing/pyproject.toml index 1060e928..3aad21f2 100644 --- a/packages/aws-durable-execution-sdk-python-testing/pyproject.toml +++ b/packages/aws-durable-execution-sdk-python-testing/pyproject.toml @@ -105,7 +105,10 @@ target-version = "py311" [tool.ruff.lint] preview = true -select = ["E4", "E7", "E9", "F", "TID252"] # pycodestyle (E4/E7/E9) + Pyflakes + absolute imports +select = ["E4", "E7", "E9", "F", "TID251", "TID252"] # pycodestyle (E4/E7/E9) + Pyflakes + banned APIs + absolute imports + +[tool.ruff.lint.flake8-tidy-imports.banned-api] +"datetime.datetime.now".msg = "Read time from the runner Clock (clock.now()) or clock.real_now() so all timestamps share one clock domain." [tool.ruff.lint.isort] known-first-party = ["aws_durable_execution_sdk_python_testing"] @@ -114,6 +117,7 @@ lines-after-imports = 2 [tool.ruff.lint.per-file-ignores] "tests/**" = [ + "TID251", "ARG001", "ARG002", "ARG005", diff --git a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/core.py b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/core.py index e1305560..89912b97 100644 --- a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/core.py +++ b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/core.py @@ -20,6 +20,8 @@ from aws_durable_execution_sdk_python_testing.token import CheckpointToken if TYPE_CHECKING: + from datetime import datetime + from aws_durable_execution_sdk_python.lambda_service import ( Operation, OperationUpdate, @@ -79,6 +81,7 @@ def apply( updates: list[OperationUpdate], client_token: str | None, dispatcher: CheckpointRequestDispatcher, + now: datetime, ) -> CheckpointResult: """Apply ``updates`` to ``execution`` and compute the response delta. @@ -88,6 +91,10 @@ def apply( idempotency entry for a byte-identical replay of a retried call. The caller is responsible for the invocation gate, locking, persistence, and applying the returned effects. + + ``now`` is the checkpoint's single source of "now", resolved from + the execution's clock, so all timestamps stamped by this apply + advance with modeled time under a skip clock. """ effects: list[CheckpointEffect] = [] if updates: @@ -97,6 +104,7 @@ def apply( updates=updates, client_token=client_token, touch=execution.touch_operation, + now=now, ) new_token_sequence: int = execution.advance_token_sequence() diff --git a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processor.py b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processor.py index c9565b5c..d9360e0a 100644 --- a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processor.py +++ b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processor.py @@ -20,6 +20,7 @@ from aws_durable_execution_sdk_python_testing.checkpoint.transformer import ( CheckpointRequestDispatcher, ) +from aws_durable_execution_sdk_python_testing.clock import RealClock from aws_durable_execution_sdk_python_testing.exceptions import ( InvalidParameterValueException, ) @@ -30,6 +31,7 @@ if TYPE_CHECKING: from aws_durable_execution_sdk_python.lambda_service import OperationUpdate + from aws_durable_execution_sdk_python_testing.clock import Clock from aws_durable_execution_sdk_python_testing.execution import Execution from aws_durable_execution_sdk_python_testing.observer import ExecutionObserver from aws_durable_execution_sdk_python_testing.scheduler import Scheduler @@ -52,10 +54,15 @@ def __init__( self, store: ExecutionStore, scheduler: Scheduler, # noqa: ARG002 — kept for backward-compatible signature + clock: Clock | None = None, ): self._store = store self._observers: list[ExecutionObserver] = [] self._dispatcher = CheckpointRequestDispatcher() + # The runner's clock, shared with the Executor so stamping uses + # the same "now" as timer arming. Defaults to a real clock for + # standalone construction (e.g. direct unit tests). + self._clock: Clock = clock if clock is not None else RealClock() def add_execution_observer(self, observer: ExecutionObserver) -> None: """Add observer for execution events.""" @@ -91,12 +98,14 @@ def process_checkpoint( msg = "Invalid checkpoint token" raise InvalidParameterValueException(msg) + now = self._clock.now() result = CheckpointCore.apply( execution, checkpoint_token, updates, client_token, self._dispatcher, + now, ) self._store.update(execution) diff --git a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processors/base.py b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processors/base.py index 56933d5b..52eb64c2 100644 --- a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processors/base.py +++ b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processors/base.py @@ -34,22 +34,29 @@ def process( current_op: Operation | None, notifier: ExecutionNotifier, execution_arn: str, + now: datetime.datetime, ) -> Operation | None: - """Process an operation update and return the transformed operation.""" + """Process an operation update and return the transformed operation. + + ``now`` is the checkpoint's single source of "now", resolved from + the execution's clock so timestamps advance with modeled time + under a skip clock and match wall-clock under a real clock. + """ raise NotImplementedError def _get_start_time( - self, current_operation: Operation | None + self, current_operation: Operation | None, now: datetime.datetime ) -> datetime.datetime | None: start_time: datetime.datetime | None = ( - current_operation.start_timestamp - if current_operation - else datetime.datetime.now(tz=datetime.UTC) + current_operation.start_timestamp if current_operation else now ) return start_time def _get_end_time( - self, current_operation: Operation | None, status: OperationStatus + self, + current_operation: Operation | None, + status: OperationStatus, + now: datetime.datetime, ) -> datetime.datetime | None: """Get end timestamp for operation based on current state and status.""" if current_operation and current_operation.end_timestamp: @@ -61,7 +68,7 @@ def _get_end_time( OperationStatus.TIMED_OUT, OperationStatus.STOPPED, }: - return datetime.datetime.now(tz=datetime.UTC) + return now return None def _create_execution_details( @@ -151,11 +158,14 @@ def _translate_update_to_operation( update: OperationUpdate, current_operation: Operation | None, status: OperationStatus, + now: datetime.datetime, ) -> Operation: """Transform OperationUpdate to Operation, always creating new Operation.""" - start_time: datetime.datetime | None = self._get_start_time(current_operation) + start_time: datetime.datetime | None = self._get_start_time( + current_operation, now + ) end_time: datetime.datetime | None = self._get_end_time( - current_operation, status + current_operation, status, now ) execution_details = self._create_execution_details(update) @@ -163,7 +173,7 @@ def _translate_update_to_operation( step_details = self._create_step_details(update, current_operation) callback_details = self._create_callback_details(update) invoke_details = self._create_invoke_details(update) - wait_details = self._create_wait_details(update, current_operation) + wait_details = self._create_wait_details(update, current_operation, now) return Operation( operation_id=update.operation_id, @@ -183,7 +193,10 @@ def _translate_update_to_operation( ) def _create_wait_details( - self, update: OperationUpdate, current_operation: Operation | None + self, + update: OperationUpdate, + current_operation: Operation | None, + now: datetime.datetime, ) -> WaitDetails | None: """Create WaitDetails from OperationUpdate.""" if update.operation_type == OperationType.WAIT and update.wait_options: @@ -192,8 +205,8 @@ def _create_wait_details( current_operation.wait_details.scheduled_end_timestamp ) else: - scheduled_end_timestamp = datetime.datetime.now( - tz=datetime.UTC - ) + timedelta(seconds=update.wait_options.wait_seconds) + scheduled_end_timestamp = now + timedelta( + seconds=update.wait_options.wait_seconds + ) return WaitDetails(scheduled_end_timestamp=scheduled_end_timestamp) return None diff --git a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processors/callback.py b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processors/callback.py index 48c2f016..1d1ede7b 100644 --- a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processors/callback.py +++ b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processors/callback.py @@ -35,6 +35,7 @@ def process( current_op: Operation | None, notifier: ExecutionNotifier, # noqa: ARG002 execution_arn: str, # noqa: ARG002 + now: datetime.datetime, ) -> Operation: """Process CALLBACK operation update with scheduler integration for activities.""" match update.action: @@ -58,10 +59,12 @@ def process( status: OperationStatus = OperationStatus.STARTED - start_time: datetime.datetime | None = self._get_start_time(current_op) + start_time: datetime.datetime | None = self._get_start_time( + current_op, now + ) end_time: datetime.datetime | None = self._get_end_time( - current_op, status + current_op, status, now ) operation: Operation = Operation( diff --git a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processors/context.py b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processors/context.py index 182bf916..9d3e22ec 100644 --- a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processors/context.py +++ b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processors/context.py @@ -20,6 +20,8 @@ if TYPE_CHECKING: + from datetime import datetime + from aws_durable_execution_sdk_python_testing.observer import ExecutionNotifier @@ -32,6 +34,7 @@ def process( current_op: Operation | None, notifier: ExecutionNotifier, # noqa: ARG002 execution_arn: str, # noqa: ARG002 + now: datetime, ) -> Operation: """Process CONTEXT operation update for context state transitions.""" match update.action: @@ -41,18 +44,21 @@ def process( update=update, current_operation=current_op, status=OperationStatus.STARTED, + now=now, ) case OperationAction.SUCCEED: return self._translate_update_to_operation( update=update, current_operation=current_op, status=OperationStatus.SUCCEEDED, + now=now, ) case OperationAction.FAIL: return self._translate_update_to_operation( update=update, current_operation=current_op, status=OperationStatus.FAILED, + now=now, ) case _: msg: str = "Invalid action for CONTEXT operation." diff --git a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processors/execution.py b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processors/execution.py index e8ad2ef0..f8b29321 100644 --- a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processors/execution.py +++ b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processors/execution.py @@ -17,6 +17,8 @@ if TYPE_CHECKING: + from datetime import datetime + from aws_durable_execution_sdk_python_testing.observer import ExecutionNotifier @@ -29,6 +31,7 @@ def process( current_op: Operation | None, # noqa: ARG002 notifier: ExecutionNotifier, execution_arn: str, + now: datetime, # noqa: ARG002 ) -> Operation | None: """Process EXECUTION operation update for workflow completion/failure.""" match update.action: diff --git a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processors/step.py b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processors/step.py index af0d0338..26777c09 100644 --- a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processors/step.py +++ b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processors/step.py @@ -2,7 +2,7 @@ from __future__ import annotations -from datetime import UTC, datetime, timedelta +from datetime import datetime, timedelta from typing import TYPE_CHECKING from aws_durable_execution_sdk_python.lambda_service import ( @@ -34,6 +34,7 @@ def process( current_op: Operation | None, notifier: ExecutionNotifier, execution_arn: str, + now: datetime, ) -> Operation: """Process STEP operation update with scheduler integration for retries.""" match update.action: @@ -42,6 +43,7 @@ def process( update=update, current_operation=current_op, status=OperationStatus.STARTED, + now=now, ) case OperationAction.RETRY: # set Status=PENDING, next attempt time, attempt count + 1 @@ -50,7 +52,7 @@ def process( if update.step_options else 0 ) - next_attempt_time = datetime.now(UTC) + timedelta(seconds=delay) + next_attempt_time = now + timedelta(seconds=delay) # Build new step_details with incremented attempt current_attempt = ( @@ -72,9 +74,7 @@ def process( status=OperationStatus.PENDING, parent_id=update.parent_id, name=update.name, - start_timestamp=( - current_op.start_timestamp if current_op else datetime.now(UTC) - ), + start_timestamp=(current_op.start_timestamp if current_op else now), end_timestamp=None, sub_type=update.sub_type, execution_details=current_op.execution_details @@ -103,12 +103,14 @@ def process( update=update, current_operation=current_op, status=OperationStatus.SUCCEEDED, + now=now, ) case OperationAction.FAIL: return self._translate_update_to_operation( update=update, current_operation=current_op, status=OperationStatus.FAILED, + now=now, ) case _: msg: str = "Invalid action for STEP operation." diff --git a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processors/wait.py b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processors/wait.py index b84387e6..bac02adb 100644 --- a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processors/wait.py +++ b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processors/wait.py @@ -2,9 +2,7 @@ from __future__ import annotations -import logging -import os -from datetime import UTC, datetime, timedelta +from datetime import datetime, timedelta from typing import TYPE_CHECKING from aws_durable_execution_sdk_python.lambda_service import ( @@ -36,6 +34,7 @@ def process( current_op: Operation | None, notifier: ExecutionNotifier, execution_arn: str, + now: datetime, ) -> Operation: """Process WAIT operation update with scheduler integration for timers.""" match update.action: @@ -43,13 +42,8 @@ def process( wait_seconds = ( update.wait_options.wait_seconds if update.wait_options else 0 ) - time_scale = float(os.getenv("DURABLE_EXECUTION_TIME_SCALE", "1.0")) - logging.info("Using DURABLE_EXECUTION_TIME_SCALE: %f", time_scale) - scaled_wait_seconds = wait_seconds * time_scale - scheduled_end_timestamp = datetime.now(UTC) + timedelta( - seconds=scaled_wait_seconds - ) + scheduled_end_timestamp = now + timedelta(seconds=wait_seconds) # Create WaitDetails with scheduled timestamp wait_details = WaitDetails( @@ -63,7 +57,7 @@ def process( status=OperationStatus.STARTED, parent_id=update.parent_id, name=update.name, - start_timestamp=datetime.now(UTC), + start_timestamp=now, end_timestamp=None, sub_type=update.sub_type, execution_details=None, @@ -87,6 +81,7 @@ def process( update=update, current_operation=current_op, status=OperationStatus.CANCELLED, + now=now, ) case _: msg: str = "Invalid action for WAIT operation." diff --git a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/transformer.py b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/transformer.py index 9b344334..62971fd6 100644 --- a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/transformer.py +++ b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/transformer.py @@ -8,7 +8,6 @@ from __future__ import annotations import json -from datetime import UTC, datetime from typing import TYPE_CHECKING, ClassVar from aws_durable_execution_sdk_python.lambda_service import ( @@ -38,6 +37,7 @@ if TYPE_CHECKING: from collections.abc import Callable, MutableMapping + from datetime import datetime from aws_durable_execution_sdk_python.lambda_service import ( OperationUpdate, @@ -83,6 +83,7 @@ def apply_updates( updates: list[OperationUpdate], client_token: str | None, # noqa: ARG002 — reserved for future idempotency diagnostics touch: Callable[[str], None], + now: datetime, ) -> list[CheckpointEffect]: """Apply ``updates`` to ``execution`` in place. @@ -124,6 +125,7 @@ def apply_updates( current_op=current_op, notifier=collector, execution_arn=execution.durable_execution_arn, + now=now, ) if updated_op is None: continue @@ -143,7 +145,7 @@ def apply_updates( touch(update.operation_id) execution.updates.extend(updates) - execution.update_timestamps.extend(datetime.now(UTC) for _ in updates) + execution.update_timestamps.extend(now for _ in updates) return collector.effects diff --git a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/cli.py b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/cli.py index d9452241..fce3d6ff 100644 --- a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/cli.py +++ b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/cli.py @@ -221,6 +221,12 @@ def _create_start_server_parser(self, subparsers) -> None: default=900, help="Per-invocation timeout in seconds, simulates Lambda Timeout (default: 900)", ) + start_server_parser.add_argument( + "--skip-time", + action=argparse.BooleanOptionalAction, + default=False, + help="Skip durable timer wall-clock waits; history keeps real modeled durations. Default is real timing (--no-skip-time); pass --skip-time to opt in", + ) start_server_parser.set_defaults(func=self.start_server_command) def _create_invoke_parser(self, subparsers) -> None: @@ -294,6 +300,7 @@ def start_server_command(self, args: argparse.Namespace) -> int: store_type=StoreType(args.store_type), store_path=args.store_path, invocation_timeout_seconds=args.invocation_timeout, + skip_time=args.skip_time, ) logger.info( diff --git a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/client.py b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/client.py index f27a98e6..38dbf701 100644 --- a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/client.py +++ b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/client.py @@ -9,6 +9,7 @@ StateOutput, ) +from aws_durable_execution_sdk_python_testing.clock import real_now from aws_durable_execution_sdk_python_testing.checkpoint.processor import ( CheckpointProcessor, ) @@ -57,4 +58,4 @@ def get_execution_state( def stop(self, execution_arn: str, payload: bytes | None) -> datetime.datetime: # noqa: ARG002 # TODO: implement # Return current time for in-memory testing - return datetime.datetime.now(tz=datetime.UTC) + return real_now() diff --git a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/clock.py b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/clock.py new file mode 100644 index 00000000..0a9bfb3d --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/clock.py @@ -0,0 +1,69 @@ +"""Clock sources for scheduler-driven durable timers. + +A clock decides how long the local runner actually waits before a +scheduled wake fires, and supplies the "now" used to stamp operation +and invocation timestamps. All time reads in the runner go through one +clock so history events share a single, totally ordered clock domain. +""" + +from __future__ import annotations + +import datetime +import threading +from typing import Protocol + + +def real_now() -> datetime.datetime: + """Return the current wall-clock time in UTC.""" + return datetime.datetime.now(datetime.UTC) # noqa: TID251 + + +class Clock(Protocol): + """Time source used to arm timers and stamp timestamps.""" + + def now(self) -> datetime.datetime: + """Return the current time on this clock.""" + ... + + def arm(self, wake: datetime.datetime) -> float: + """Return the real seconds to wait before firing at ``wake``.""" + ... + + +class RealClock: + """Wall-clock time source. Timers wait the real modeled duration.""" + + def now(self) -> datetime.datetime: + return real_now() + + def arm(self, wake: datetime.datetime) -> float: + return max((wake - self.now()).total_seconds(), 0.0) + + +class SkipClock: + """Virtual clock that skips over armed timer horizons. + + Tracks the total duration skipped so far: ``now`` returns real time + plus the accumulated skip, so the clock advances with real time + between jumps and every read is distinct and monotonic. Arming a + wake beyond virtual now adds the remaining gap to the accumulated + skip and returns a zero delay, so the wake fires on the next + event-loop turn rather than after the modeled duration. Guards its + state with a lock because ``arm`` and ``now`` run on different + threads. + """ + + def __init__(self) -> None: + self._skipped: datetime.timedelta = datetime.timedelta(0) + self._lock: threading.Lock = threading.Lock() + + def now(self) -> datetime.datetime: + with self._lock: + return real_now() + self._skipped + + def arm(self, wake: datetime.datetime) -> float: + with self._lock: + gap: datetime.timedelta = wake - (real_now() + self._skipped) + if gap > datetime.timedelta(0): + self._skipped += gap + return 0.0 diff --git a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/execution.py b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/execution.py index f6d06ec2..b03b45f3 100644 --- a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/execution.py +++ b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/execution.py @@ -1,7 +1,7 @@ from __future__ import annotations from dataclasses import dataclass, replace -from datetime import UTC, datetime +from datetime import datetime from enum import Enum from threading import Lock from typing import Any @@ -20,6 +20,7 @@ OperationUpdate, ) +from aws_durable_execution_sdk_python_testing.clock import real_now from aws_durable_execution_sdk_python_testing.exceptions import ( IllegalStateException, InvalidParameterValueException, @@ -284,7 +285,7 @@ def from_json_dict(cls, data: dict[str, Any]) -> Execution: return execution - def start(self) -> None: + def start(self, now: datetime | None = None) -> None: if self.start_input.invocation_id is None: msg: str = "invocation_id is required" raise InvalidParameterValueException(msg) @@ -294,7 +295,7 @@ def start(self) -> None: operation_id=self.start_input.invocation_id, parent_id=None, name=self.start_input.execution_name, - start_timestamp=datetime.now(UTC), + start_timestamp=now if now is not None else real_now(), operation_type=OperationType.EXECUTION, status=OperationStatus.STARTED, execution_details=ExecutionDetails( @@ -381,41 +382,41 @@ def _record_updated_operation(self, operation_id: str) -> None: if operation_id not in self.updated_operation_ids: self.updated_operation_ids.append(operation_id) - def complete_success(self, result: str | None) -> None: + def complete_success(self, result: str | None, now: datetime | None = None) -> None: """Complete execution successfully (DecisionType.COMPLETE_WORKFLOW_EXECUTION).""" self.result = DurableExecutionInvocationOutput( status=InvocationStatus.SUCCEEDED, result=result ) self.is_complete = True self.close_status = ExecutionStatus.SUCCEEDED - self._end_execution(OperationStatus.SUCCEEDED) + self._end_execution(OperationStatus.SUCCEEDED, now) - def complete_fail(self, error: ErrorObject) -> None: + def complete_fail(self, error: ErrorObject, now: datetime | None = None) -> None: """Complete execution with failure (DecisionType.FAIL_WORKFLOW_EXECUTION).""" self.result = DurableExecutionInvocationOutput( status=InvocationStatus.FAILED, error=error ) self.is_complete = True self.close_status = ExecutionStatus.FAILED - self._end_execution(OperationStatus.FAILED) + self._end_execution(OperationStatus.FAILED, now) - def complete_timeout(self, error: ErrorObject) -> None: + def complete_timeout(self, error: ErrorObject, now: datetime | None = None) -> None: """Complete execution with timeout.""" self.result = DurableExecutionInvocationOutput( status=InvocationStatus.FAILED, error=error ) self.is_complete = True self.close_status = ExecutionStatus.TIMED_OUT - self._end_execution(OperationStatus.TIMED_OUT) + self._end_execution(OperationStatus.TIMED_OUT, now) - def complete_stopped(self, error: ErrorObject) -> None: + def complete_stopped(self, error: ErrorObject, now: datetime | None = None) -> None: """Complete execution as terminated (TerminateWorkflowExecutionV2Request).""" self.result = DurableExecutionInvocationOutput( status=InvocationStatus.FAILED, error=error ) self.is_complete = True self.close_status = ExecutionStatus.STOPPED - self._end_execution(OperationStatus.STOPPED) + self._end_execution(OperationStatus.STOPPED, now) def find_operation(self, operation_id: str) -> tuple[int, Operation]: """Find operation by ID, return index and operation.""" @@ -437,7 +438,9 @@ def find_callback_operation(self, callback_id: str) -> tuple[int, Operation]: msg: str = f"Callback operation with callback_id [{callback_id}] not found" raise IllegalStateException(msg) - def complete_wait(self, operation_id: str) -> Operation: + def complete_wait( + self, operation_id: str, now: datetime | None = None + ) -> Operation: """Complete WAIT operation when timer fires.""" index, operation = self.find_operation(operation_id) @@ -458,7 +461,7 @@ def complete_wait(self, operation_id: str) -> Operation: self.operations[index] = replace( operation, status=OperationStatus.SUCCEEDED, - end_timestamp=datetime.now(UTC), + end_timestamp=now if now is not None else real_now(), ) self._record_updated_operation(operation_id) return self.operations[index] @@ -498,7 +501,10 @@ def complete_retry(self, operation_id: str) -> Operation: return updated_operation def complete_callback_success( - self, callback_id: str, result: bytes | None = None + self, + callback_id: str, + result: bytes | None = None, + now: datetime | None = None, ) -> Operation: """Complete CALLBACK operation with success.""" index, operation = self.find_callback_operation(callback_id) @@ -518,13 +524,13 @@ def complete_callback_success( self.operations[index] = replace( operation, status=OperationStatus.SUCCEEDED, - end_timestamp=datetime.now(UTC), + end_timestamp=now if now is not None else real_now(), callback_details=updated_callback_details, ) return self.operations[index] def complete_callback_failure( - self, callback_id: str, error: ErrorObject + self, callback_id: str, error: ErrorObject, now: datetime | None = None ) -> Operation: """Complete CALLBACK operation with failure.""" index, operation = self.find_callback_operation(callback_id) @@ -544,13 +550,13 @@ def complete_callback_failure( self.operations[index] = replace( operation, status=OperationStatus.FAILED, - end_timestamp=datetime.now(UTC), + end_timestamp=now if now is not None else real_now(), callback_details=updated_callback_details, ) return self.operations[index] def complete_callback_timeout( - self, callback_id: str, error: ErrorObject + self, callback_id: str, error: ErrorObject, now: datetime | None = None ) -> Operation: """Complete CALLBACK operation with timeout.""" index, operation = self.find_callback_operation(callback_id) @@ -570,12 +576,14 @@ def complete_callback_timeout( self.operations[index] = replace( operation, status=OperationStatus.TIMED_OUT, - end_timestamp=datetime.now(UTC), + end_timestamp=now if now is not None else real_now(), callback_details=updated_callback_details, ) return self.operations[index] - def _end_execution(self, status: OperationStatus) -> None: + def _end_execution( + self, status: OperationStatus, now: datetime | None = None + ) -> None: """Set the end_timestamp on the main EXECUTION operation when execution completes.""" execution_op: Operation = self.get_operation_execution_started() if execution_op.operation_type == OperationType.EXECUTION: @@ -589,7 +597,7 @@ def _end_execution(self, status: OperationStatus) -> None: self.operations[0] = replace( execution_op, status=status, - end_timestamp=datetime.now(UTC), + end_timestamp=now if now is not None else real_now(), ) diff --git a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/executor.py b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/executor.py index 6a61867d..aa43add7 100644 --- a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/executor.py +++ b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/executor.py @@ -6,7 +6,7 @@ import logging import threading import uuid -from datetime import UTC, datetime +from datetime import datetime from typing import TYPE_CHECKING from aws_durable_execution_sdk_python.execution import ( @@ -30,6 +30,7 @@ from aws_durable_execution_sdk_python_testing.checkpoint.processor import ( DEFAULT_MAX_INVOCATION_PAGE_BYTES, ) +from aws_durable_execution_sdk_python_testing.clock import Clock, RealClock from aws_durable_execution_sdk_python_testing.checkpoint.transformer import ( CheckpointRequestDispatcher, ) @@ -114,6 +115,7 @@ def __init__( max_invocation_page_bytes: int | None = None, invocation_timeout_seconds: int = 900, registry: ExecutionRegistry | None = None, + clock: Clock | None = None, ): self._store = store self._scheduler = scheduler @@ -122,6 +124,7 @@ def __init__( self._registry = ( registry if registry is not None else ExecutionRegistry(store, scheduler) ) + self._clock: Clock = clock if clock is not None else RealClock() self._invocation_timeout_seconds = invocation_timeout_seconds self._dispatcher = CheckpointRequestDispatcher() self._max_invocation_page_bytes = ( @@ -164,7 +167,7 @@ def start_execution( ) execution = Execution.new(input=input) - execution.start() + execution.start(now=self._clock.now()) self._store.save(execution) logger.debug("Created execution with ARN: %s", execution.durable_execution_arn) @@ -245,7 +248,7 @@ def get_execution_details(self, execution_arn: str) -> GetDurableExecutionRespon status=status, start_timestamp=execution_op.start_timestamp if execution_op.start_timestamp - else datetime.now(UTC), + else self._clock.now(), input_payload=execution_op.execution_details.input_payload if execution_op.execution_details else None, @@ -388,7 +391,7 @@ def _apply_stop( if execution.is_complete: # Idempotent: return the existing stop timestamp execution_op = execution.get_operation_execution_started() - stop_timestamp = execution_op.end_timestamp or datetime.now(UTC) + stop_timestamp = execution_op.end_timestamp or self._clock.now() return StopDurableExecutionResponse(stop_timestamp=stop_timestamp) # Use provided error or create a default one @@ -398,11 +401,12 @@ def _apply_stop( # Stop sets TERMINATED close status (different from fail) logger.info("[%s] Stopping execution.", execution_arn) - execution.complete_stopped(error=stop_error) # Sets CloseStatus.TERMINATED + stop_time: datetime = self._clock.now() + execution.complete_stopped(error=stop_error, now=stop_time) self._store.update(execution) self._complete_events(execution_arn=execution_arn) - return StopDurableExecutionResponse(stop_timestamp=datetime.now(UTC)) + return StopDurableExecutionResponse(stop_timestamp=stop_time) def get_execution_state( self, @@ -558,7 +562,7 @@ def get_execution_history( } for idx, update in enumerate(updates): ts: datetime = ( - timestamps[idx] if idx < len(timestamps) else datetime.now(UTC) + timestamps[idx] if idx < len(timestamps) else self._clock.now() ) real_op: Operation | None = ops_by_id.get(update.operation_id) @@ -847,12 +851,14 @@ def _checkpoint_execution( msg = "Invalid checkpoint token" raise InvalidParameterValueException(msg) + now = self._clock.now() result = CheckpointCore.apply( execution, checkpoint_token, updates or [], client_token, self._dispatcher, + now, ) self._store.update(execution) @@ -1000,8 +1006,7 @@ def _schedule_earliest_pending(self, execution_arn: str) -> None: existing.cancel() return - now = datetime.now(UTC) - delay = max((earliest - now).total_seconds(), 0.0) + delay = self._clock.arm(earliest) completion_event = self._completion_events.get(execution_arn) # Cancel-then-arm atomically under the supervisor lock so @@ -1031,7 +1036,7 @@ def _fire_due_operations(self, execution_arn: str) -> bool: if execution.is_complete: return False - now = datetime.now(UTC) + now = self._clock.now() completed_any = False for op in list(execution.operations): if ( @@ -1042,7 +1047,7 @@ def _fire_due_operations(self, execution_arn: str) -> bool: and op.wait_details.scheduled_end_timestamp <= now ): try: - execution.complete_wait(op.operation_id) + execution.complete_wait(op.operation_id, now=now) completed_any = True except Exception: # noqa: BLE001 logger.exception( @@ -1146,7 +1151,7 @@ def _complete_callback_success( ) -> None: callback_token = CallbackToken.from_str(callback_id) execution = self.get_execution(callback_token.execution_arn) - execution.complete_callback_success(callback_id, result) + execution.complete_callback_success(callback_id, result, now=self._clock.now()) self._store.update(execution) self._cleanup_callback_timeouts(callback_id) @@ -1202,7 +1207,9 @@ def _complete_callback_failure( ) -> None: callback_token = CallbackToken.from_str(callback_id) execution = self.get_execution(callback_token.execution_arn) - execution.complete_callback_failure(callback_id, callback_error) + execution.complete_callback_failure( + callback_id, callback_error, now=self._clock.now() + ) self._store.update(execution) self._cleanup_callback_timeouts(callback_id) @@ -1483,7 +1490,7 @@ async def invoke() -> None: return execution, invocation_input = claim - invocation_start = datetime.now(UTC) + invocation_start = self._clock.now() invoke_response = await asyncio.wait_for( asyncio.to_thread( self._invoker.invoke, @@ -1493,7 +1500,7 @@ async def invoke() -> None: ), timeout=self._invocation_timeout_seconds, ) - invocation_end = datetime.now(UTC) + invocation_end = self._clock.now() await asyncio.to_thread( lambda: self._registry.submit( execution_arn, @@ -1526,7 +1533,7 @@ async def invoke() -> None: # Invocation killed by Lambda timeout. Step operations # stay in their current state (STARTED) — no checkpoint # was sent. Record the failed invocation and re-invoke. - invocation_end = datetime.now(UTC) + invocation_end = self._clock.now() logger.warning( "[%s] Invocation timed out after %ds", execution_arn, @@ -1662,7 +1669,7 @@ def complete_execution(self, execution_arn: str, result: str | None = None) -> N """Complete execution successfully (COMPLETE_WORKFLOW_EXECUTION decision).""" logger.debug("[%s] Completing execution with result: %s", execution_arn, result) execution: Execution = self._store.load(execution_arn=execution_arn) - execution.complete_success(result=result) # Sets CloseStatus.COMPLETED + execution.complete_success(result=result, now=self._clock.now()) self._store.update(execution) if execution.result is None: msg: str = "Execution result is required" @@ -1673,7 +1680,7 @@ def fail_execution(self, execution_arn: str, error: ErrorObject) -> None: """Fail execution with error (FAIL_WORKFLOW_EXECUTION decision).""" logger.error("[%s] Completing execution with error: %s", execution_arn, error) execution: Execution = self._store.load(execution_arn=execution_arn) - execution.complete_fail(error=error) # Sets CloseStatus.FAILED + execution.complete_fail(error=error, now=self._clock.now()) self._store.update(execution) # set by complete_fail if execution.result is None: @@ -1701,7 +1708,7 @@ def on_timed_out(self, execution_arn: str, error: ErrorObject) -> None: def _apply_timeout(self, execution_arn: str, error: ErrorObject) -> None: execution = self._store.load(execution_arn=execution_arn) - execution.complete_timeout(error=error) # Sets CloseStatus.TIMED_OUT + execution.complete_timeout(error=error, now=self._clock.now()) self._store.update(execution) def on_stopped(self, execution_arn: str, error: ErrorObject) -> None: @@ -1836,7 +1843,7 @@ def _apply_callback_timeout(self, callback_id: str, error: ErrorObject) -> None: execution = self.get_execution(callback_token.execution_arn) if execution.is_complete: return - execution.complete_callback_timeout(callback_id, error) + execution.complete_callback_timeout(callback_id, error, now=self._clock.now()) self._store.update(execution) def _on_callback_timeout(self, execution_arn: str, callback_id: str) -> None: diff --git a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/model.py b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/model.py index 12e69a8c..173fcf77 100644 --- a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/model.py +++ b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/model.py @@ -3,6 +3,8 @@ from __future__ import annotations import datetime + +from aws_durable_execution_sdk_python_testing.clock import real_now import json from dataclasses import dataclass, replace from enum import Enum @@ -35,8 +37,6 @@ from aws_durable_execution_sdk_python.types import ( LambdaContext as LambdaContextProtocol, ) -from dateutil.tz import UTC - from aws_durable_execution_sdk_python_testing.exceptions import ( InvalidParameterValueException, ) @@ -328,7 +328,7 @@ def from_execution(cls, execution, status: str) -> Execution: status=status, start_timestamp=execution_op.start_timestamp if execution_op.start_timestamp - else datetime.datetime.now(datetime.UTC), + else real_now(), end_timestamp=execution_op.end_timestamp if execution_op.end_timestamp else None, @@ -1335,7 +1335,7 @@ def start_timestamp(self) -> datetime.datetime: return ( self.operation.start_timestamp if self.operation.start_timestamp is not None - else datetime.datetime.now(UTC) + else real_now() ) @property @@ -1343,7 +1343,7 @@ def end_timestamp(self) -> datetime.datetime: return ( self.operation.end_timestamp if self.operation.end_timestamp is not None - else datetime.datetime.now(UTC) + else real_now() ) @@ -2664,7 +2664,7 @@ def events_to_operations(events: list[Event]) -> list[Operation]: name=event.name, parent_id=event.parent_id, sub_type=sub_type, - start_timestamp=datetime.datetime.now(tz=datetime.timezone.utc), + start_timestamp=event.event_timestamp, ) # Merge with previous operation if it exists diff --git a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/runner.py b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/runner.py index 4b3afac8..b70cbd00 100644 --- a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/runner.py +++ b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/runner.py @@ -36,6 +36,7 @@ DEFAULT_MAX_INVOCATION_PAGE_BYTES, ) from aws_durable_execution_sdk_python_testing.client import InMemoryServiceClient +from aws_durable_execution_sdk_python_testing.clock import Clock, RealClock, SkipClock from aws_durable_execution_sdk_python_testing.worker.registry import ExecutionRegistry from aws_durable_execution_sdk_python_testing.exceptions import ( DurableFunctionsLocalRunnerError, @@ -111,6 +112,12 @@ class WebRunnerConfig: # Timeout configuration invocation_timeout_seconds: int = 900 + # Skip durable timer wall-clock waits (waits and step retries complete + # on the next scheduler turn while history keeps the real modeled + # durations). Defaults False so the server emulates real cloud timing; + # set True to opt into fast skip. + skip_time: bool = False + # Invocation-input pagination cap (bytes). Handler receives pages # up to this size; larger state is served via # GetDurableExecutionState. None falls back to Executor default @@ -603,6 +610,7 @@ def __init__( execution_timeout: int = 300, invocation_timeout: int = 900, store: ExecutionStore | None = None, + skip_time: bool = True, # noqa: FBT001, FBT002 ): self._execution_timeout = execution_timeout self._invocation_timeout = invocation_timeout @@ -619,11 +627,13 @@ def __init__( store if store is not None else InMemoryExecutionStore() ) self.poll_interval = poll_interval + self._clock: Clock = SkipClock() if skip_time else RealClock() + self._registry = ExecutionRegistry(self._store, self._scheduler) self._checkpoint_processor = CheckpointProcessor( store=self._store, scheduler=self._scheduler, + clock=self._clock, ) - self._registry = ExecutionRegistry(self._store, self._scheduler) self._service_client = InMemoryServiceClient( self._checkpoint_processor, self._registry ) @@ -640,6 +650,7 @@ def __init__( max_invocation_page_bytes=self._max_invocation_page_bytes, invocation_timeout_seconds=invocation_timeout, registry=self._registry, + clock=self._clock, ) # Wire up observer pattern - CheckpointProcessor uses this to notify executor of state changes @@ -870,8 +881,14 @@ def start(self) -> None: ) # Create shared CheckpointProcessor - checkpoint_processor = CheckpointProcessor(self._store, self._scheduler) + clock: Clock = SkipClock() if self._config.skip_time else RealClock() + self._clock = clock self._registry = ExecutionRegistry(self._store, self._scheduler) + checkpoint_processor = CheckpointProcessor( + self._store, + self._scheduler, + clock=clock, + ) # Create executor with all dependencies including checkpoint processor self._executor = Executor( @@ -882,6 +899,7 @@ def start(self) -> None: max_invocation_page_bytes=resolved_max_page_bytes, invocation_timeout_seconds=self._config.invocation_timeout_seconds, registry=self._registry, + clock=clock, ) # Add executor as observer to the checkpoint processor diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/checkpoint/processors/base_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/checkpoint/processors/base_test.py index 5f92b58e..3d6e739e 100644 --- a/packages/aws-durable-execution-sdk-python-testing/tests/checkpoint/processors/base_test.py +++ b/packages/aws-durable-execution-sdk-python-testing/tests/checkpoint/processors/base_test.py @@ -37,7 +37,9 @@ def test_process_not_implemented(): ) try: - processor.process(update, None, Mock(), "test-arn") + processor.process( + update, None, Mock(), "test-arn", datetime.datetime.now(tz=datetime.UTC) + ) pytest.fail("Expected NotImplementedError") except NotImplementedError: pass @@ -46,18 +48,25 @@ def test_process_not_implemented(): class MockProcessor(OperationProcessor): """Mock processor for testing base functionality.""" - def process(self, update, current_op, notifier, execution_arn): + def process(self, update, current_op, notifier, execution_arn, now=None): return self._translate_update_to_operation( - update, current_op, OperationStatus.STARTED + update, + current_op, + OperationStatus.STARTED, + now or datetime.datetime.now(tz=datetime.UTC), ) - def translate_update(self, update, current_op, status): + def translate_update(self, update, current_op, status, now=None): """Public method to access _translate_update_to_operation for testing.""" - return self._translate_update_to_operation(update, current_op, status) + return self._translate_update_to_operation( + update, current_op, status, now or datetime.datetime.now(tz=datetime.UTC) + ) - def get_end_time(self, current_op, status): + def get_end_time(self, current_op, status, now=None): """Public method to access _get_end_time for testing.""" - return self._get_end_time(current_op, status) + return self._get_end_time( + current_op, status, now or datetime.datetime.now(tz=datetime.UTC) + ) def create_execution_details(self, update): """Public method to access _create_execution_details for testing.""" @@ -79,9 +88,11 @@ def create_invoke_details(self, update): """Public method to access _create_invoke_details for testing.""" return self._create_invoke_details(update) - def create_wait_details(self, update, current_op): + def create_wait_details(self, update, current_op, now=None): """Public method to access _create_wait_details for testing.""" - return self._create_wait_details(update, current_op) + return self._create_wait_details( + update, current_op, now or datetime.datetime.now(tz=datetime.UTC) + ) def test_get_end_time_with_existing_end_timestamp(): diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/checkpoint/processors/callback_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/checkpoint/processors/callback_test.py index 24bc1297..fdf01629 100644 --- a/packages/aws-durable-execution-sdk-python-testing/tests/checkpoint/processors/callback_test.py +++ b/packages/aws-durable-execution-sdk-python-testing/tests/checkpoint/processors/callback_test.py @@ -1,5 +1,6 @@ """Tests for callback operation processor.""" +from datetime import UTC, datetime from unittest.mock import Mock import pytest @@ -55,7 +56,11 @@ def test_process_start_action(): ) result = processor.process( - update, None, notifier, "arn:aws:states:us-east-1:123456789012:execution:test" + update, + None, + notifier, + "arn:aws:states:us-east-1:123456789012:execution:test", + datetime.now(UTC), ) assert isinstance(result, Operation) @@ -85,6 +90,7 @@ def test_process_start_action_with_current_operation(): current_op, notifier, "arn:aws:states:us-east-1:123456789012:execution:test", + datetime.now(UTC), ) assert isinstance(result, Operation) @@ -112,6 +118,7 @@ def test_process_invalid_action(): None, notifier, "arn:aws:states:us-east-1:123456789012:execution:test", + datetime.now(UTC), ) @@ -134,6 +141,7 @@ def test_process_fail_action(): None, notifier, "arn:aws:states:us-east-1:123456789012:execution:test", + datetime.now(UTC), ) @@ -156,6 +164,7 @@ def test_process_cancel_action(): None, notifier, "arn:aws:states:us-east-1:123456789012:execution:test", + datetime.now(UTC), ) @@ -178,6 +187,7 @@ def test_process_retry_action(): None, notifier, "arn:aws:states:us-east-1:123456789012:execution:test", + datetime.now(UTC), ) @@ -194,7 +204,11 @@ def test_process_with_payload(): ) result = processor.process( - update, None, notifier, "arn:aws:states:us-east-1:123456789012:execution:test" + update, + None, + notifier, + "arn:aws:states:us-east-1:123456789012:execution:test", + datetime.now(UTC), ) assert result.callback_details.result == "test-payload" @@ -213,7 +227,11 @@ def test_process_with_parent_id(): ) result = processor.process( - update, None, notifier, "arn:aws:states:us-east-1:123456789012:execution:test" + update, + None, + notifier, + "arn:aws:states:us-east-1:123456789012:execution:test", + datetime.now(UTC), ) assert result.parent_id == "parent-456" @@ -232,7 +250,11 @@ def test_process_with_sub_type(): ) result = processor.process( - update, None, notifier, "arn:aws:states:us-east-1:123456789012:execution:test" + update, + None, + notifier, + "arn:aws:states:us-east-1:123456789012:execution:test", + datetime.now(UTC), ) assert result.sub_type == "activity" @@ -250,7 +272,11 @@ def test_notifier_not_called_for_start(): ) processor.process( - update, None, notifier, "arn:aws:states:us-east-1:123456789012:execution:test" + update, + None, + notifier, + "arn:aws:states:us-east-1:123456789012:execution:test", + datetime.now(UTC), ) assert len(notifier.completed_calls) == 0 diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/checkpoint/processors/context_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/checkpoint/processors/context_test.py index a070bc17..77b73467 100644 --- a/packages/aws-durable-execution-sdk-python-testing/tests/checkpoint/processors/context_test.py +++ b/packages/aws-durable-execution-sdk-python-testing/tests/checkpoint/processors/context_test.py @@ -57,7 +57,7 @@ def test_process_start_action(): name="test-context", ) - result = processor.process(update, None, notifier, execution_arn) + result = processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert isinstance(result, Operation) assert result.operation_id == "context-123" @@ -82,7 +82,9 @@ def test_process_start_action_with_current_operation(): name="test-context", ) - result = processor.process(update, current_op, notifier, execution_arn) + result = processor.process( + update, current_op, notifier, execution_arn, datetime.now(UTC) + ) assert result.start_timestamp == current_op.start_timestamp assert result.status == OperationStatus.STARTED @@ -101,7 +103,7 @@ def test_process_succeed_action(): payload="success-result", ) - result = processor.process(update, None, notifier, execution_arn) + result = processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert isinstance(result, Operation) assert result.operation_id == "context-123" @@ -126,7 +128,9 @@ def test_process_succeed_action_with_current_operation(): payload="success-result", ) - result = processor.process(update, current_op, notifier, execution_arn) + result = processor.process( + update, current_op, notifier, execution_arn, datetime.now(UTC) + ) assert result.start_timestamp == current_op.start_timestamp assert result.status == OperationStatus.SUCCEEDED @@ -146,7 +150,7 @@ def test_process_fail_action(): error=error, ) - result = processor.process(update, None, notifier, execution_arn) + result = processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert isinstance(result, Operation) assert result.operation_id == "context-123" @@ -172,7 +176,9 @@ def test_process_fail_action_with_current_operation(): error=error, ) - result = processor.process(update, current_op, notifier, execution_arn) + result = processor.process( + update, current_op, notifier, execution_arn, datetime.now(UTC) + ) assert result.start_timestamp == current_op.start_timestamp assert result.status == OperationStatus.FAILED @@ -193,7 +199,7 @@ def test_process_fail_action_with_payload_and_error(): error=error, ) - result = processor.process(update, None, notifier, execution_arn) + result = processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert result.context_details.result == "partial-result" assert result.context_details.error == error @@ -214,7 +220,7 @@ def test_process_invalid_action(): with pytest.raises( InvalidParameterValueException, match="Invalid action for CONTEXT operation" ): - processor.process(update, None, notifier, execution_arn) + processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) def test_process_cancel_action(): @@ -232,7 +238,7 @@ def test_process_cancel_action(): with pytest.raises( InvalidParameterValueException, match="Invalid action for CONTEXT operation" ): - processor.process(update, None, notifier, execution_arn) + processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) def test_process_with_parent_id(): @@ -248,7 +254,7 @@ def test_process_with_parent_id(): parent_id="parent-456", ) - result = processor.process(update, None, notifier, execution_arn) + result = processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert result.parent_id == "parent-456" @@ -266,7 +272,7 @@ def test_process_with_sub_type(): sub_type="parallel", ) - result = processor.process(update, None, notifier, execution_arn) + result = processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert result.sub_type == "parallel" @@ -283,7 +289,7 @@ def test_process_start_without_payload(): name="test-context", ) - result = processor.process(update, None, notifier, execution_arn) + result = processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert result.context_details.result is None assert result.context_details.error is None @@ -301,7 +307,7 @@ def test_process_succeed_without_payload(): name="test-context", ) - result = processor.process(update, None, notifier, execution_arn) + result = processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert result.context_details.result is None assert result.context_details.error is None @@ -319,7 +325,7 @@ def test_process_fail_without_error(): name="test-context", ) - result = processor.process(update, None, notifier, execution_arn) + result = processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert result.context_details.result is None assert result.context_details.error is None @@ -337,7 +343,7 @@ def test_no_notifier_calls(): name="test-context", ) - processor.process(update, None, notifier, execution_arn) + processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert len(notifier.completed_calls) == 0 assert len(notifier.failed_calls) == 0 @@ -357,7 +363,7 @@ def test_end_timestamp_set_for_terminal_states(): name="test-context", ) - result = processor.process(update, None, notifier, execution_arn) + result = processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert result.end_timestamp is not None @@ -374,6 +380,6 @@ def test_end_timestamp_not_set_for_non_terminal_states(): name="test-context", ) - result = processor.process(update, None, notifier, execution_arn) + result = processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert result.end_timestamp is None diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/checkpoint/processors/execution_processor_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/checkpoint/processors/execution_processor_test.py index 37c91ea3..1ca805d3 100644 --- a/packages/aws-durable-execution-sdk-python-testing/tests/checkpoint/processors/execution_processor_test.py +++ b/packages/aws-durable-execution-sdk-python-testing/tests/checkpoint/processors/execution_processor_test.py @@ -1,5 +1,6 @@ """Tests for execution operation processor.""" +from datetime import UTC, datetime from unittest.mock import Mock from aws_durable_execution_sdk_python.lambda_service import ( @@ -50,7 +51,7 @@ def test_process_succeed_action(): payload="success-result", ) - result = processor.process(update, None, notifier, execution_arn) + result = processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert result is None assert len(notifier.completed_calls) == 1 @@ -72,7 +73,9 @@ def test_process_succeed_action_with_current_operation(): payload="success-result", ) - result = processor.process(update, current_op, notifier, execution_arn) + result = processor.process( + update, current_op, notifier, execution_arn, datetime.now(UTC) + ) assert result is None assert len(notifier.completed_calls) == 1 @@ -90,7 +93,7 @@ def test_process_succeed_action_without_payload(): action=OperationAction.SUCCEED, ) - result = processor.process(update, None, notifier, execution_arn) + result = processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert result is None assert len(notifier.completed_calls) == 1 @@ -110,7 +113,7 @@ def test_process_fail_action_with_error(): error=error, ) - result = processor.process(update, None, notifier, execution_arn) + result = processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert result is None assert len(notifier.failed_calls) == 1 @@ -129,7 +132,7 @@ def test_process_fail_action_without_error(): action=OperationAction.FAIL, ) - result = processor.process(update, None, notifier, execution_arn) + result = processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert result is None assert len(notifier.failed_calls) == 1 @@ -153,7 +156,7 @@ def test_process_start_action(): action=OperationAction.START, ) - result = processor.process(update, None, notifier, execution_arn) + result = processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert result is None assert len(notifier.failed_calls) == 1 @@ -173,7 +176,7 @@ def test_process_retry_action(): action=OperationAction.RETRY, ) - result = processor.process(update, None, notifier, execution_arn) + result = processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert result is None assert len(notifier.failed_calls) == 1 @@ -193,7 +196,7 @@ def test_process_cancel_action(): action=OperationAction.CANCEL, ) - result = processor.process(update, None, notifier, execution_arn) + result = processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert result is None assert len(notifier.failed_calls) == 1 @@ -217,7 +220,9 @@ def test_process_with_current_operation_and_error(): error=error, ) - result = processor.process(update, current_op, notifier, execution_arn) + result = processor.process( + update, current_op, notifier, execution_arn, datetime.now(UTC) + ) assert result is None assert len(notifier.failed_calls) == 1 @@ -236,7 +241,7 @@ def test_no_wait_timer_or_step_retry_calls(): payload="result", ) - processor.process(update, None, notifier, execution_arn) + processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert len(notifier.wait_timer_calls) == 0 assert len(notifier.step_retry_calls) == 0 diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/checkpoint/processors/step_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/checkpoint/processors/step_test.py index ea37b0c9..b562d6dd 100644 --- a/packages/aws-durable-execution-sdk-python-testing/tests/checkpoint/processors/step_test.py +++ b/packages/aws-durable-execution-sdk-python-testing/tests/checkpoint/processors/step_test.py @@ -59,7 +59,7 @@ def test_process_start_action(): name="test-step", ) - result = processor.process(update, None, notifier, execution_arn) + result = processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert isinstance(result, Operation) assert result.operation_id == "step-123" @@ -84,7 +84,9 @@ def test_process_start_action_with_current_operation(): name="test-step", ) - result = processor.process(update, current_op, notifier, execution_arn) + result = processor.process( + update, current_op, notifier, execution_arn, datetime.now(UTC) + ) assert result.start_timestamp == current_op.start_timestamp @@ -112,7 +114,9 @@ def test_process_retry_action(): step_options=step_options, ) - result = processor.process(update, current_op, notifier, execution_arn) + result = processor.process( + update, current_op, notifier, execution_arn, datetime.now(UTC) + ) assert isinstance(result, Operation) assert result.operation_id == "step-123" @@ -147,7 +151,9 @@ def test_process_retry_action_without_step_options(): name="test-step", ) - result = processor.process(update, current_op, notifier, execution_arn) + result = processor.process( + update, current_op, notifier, execution_arn, datetime.now(UTC) + ) assert result.step_details.attempt == 1 # no per-op notifier call; central scheduler. @@ -168,7 +174,7 @@ def test_process_retry_action_without_current_operation(): step_options=step_options, ) - result = processor.process(update, None, notifier, execution_arn) + result = processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert result.step_details.attempt == 1 assert result.step_details.result is None @@ -198,7 +204,9 @@ def test_process_retry_action_without_current_step_details(): step_options=step_options, ) - result = processor.process(update, current_op, notifier, execution_arn) + result = processor.process( + update, current_op, notifier, execution_arn, datetime.now(UTC) + ) assert result.step_details.attempt == 1 @@ -216,7 +224,7 @@ def test_process_succeed_action(): payload="success-result", ) - result = processor.process(update, None, notifier, execution_arn) + result = processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert isinstance(result, Operation) assert result.operation_id == "step-123" @@ -241,7 +249,9 @@ def test_process_succeed_action_with_current_operation(): payload="success-result", ) - result = processor.process(update, current_op, notifier, execution_arn) + result = processor.process( + update, current_op, notifier, execution_arn, datetime.now(UTC) + ) assert result.start_timestamp == current_op.start_timestamp assert result.status == OperationStatus.SUCCEEDED @@ -262,7 +272,7 @@ def test_process_fail_action(): error=error, ) - result = processor.process(update, None, notifier, execution_arn) + result = processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert isinstance(result, Operation) assert result.operation_id == "step-123" @@ -288,7 +298,9 @@ def test_process_fail_action_with_current_operation(): error=error, ) - result = processor.process(update, current_op, notifier, execution_arn) + result = processor.process( + update, current_op, notifier, execution_arn, datetime.now(UTC) + ) assert result.start_timestamp == current_op.start_timestamp assert result.status == OperationStatus.FAILED @@ -310,7 +322,7 @@ def test_process_invalid_action(): with pytest.raises( InvalidParameterValueException, match="Invalid action for STEP operation" ): - processor.process(update, None, notifier, execution_arn) + processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) def test_process_with_parent_id(): @@ -326,7 +338,7 @@ def test_process_with_parent_id(): parent_id="parent-456", ) - result = processor.process(update, None, notifier, execution_arn) + result = processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert result.parent_id == "parent-456" @@ -344,7 +356,7 @@ def test_process_with_sub_type(): sub_type="lambda", ) - result = processor.process(update, None, notifier, execution_arn) + result = processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert result.sub_type == "lambda" @@ -374,7 +386,9 @@ def test_retry_preserves_current_operation_details(): step_options=step_options, ) - result = processor.process(update, current_op, notifier, execution_arn) + result = processor.process( + update, current_op, notifier, execution_arn, datetime.now(UTC) + ) assert result.step_details.attempt == 3 assert result.step_details.result is None @@ -398,7 +412,7 @@ def test_no_completed_or_failed_calls_for_non_execution_actions(): name="test-step", ) - processor.process(update, None, notifier, execution_arn) + processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert len(notifier.completed_calls) == 0 assert len(notifier.failed_calls) == 0 @@ -417,6 +431,6 @@ def test_no_step_retry_calls_for_non_retry_actions(): name="test-step", ) - processor.process(update, None, notifier, execution_arn) + processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert len(notifier.step_retry_calls) == 0 diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/checkpoint/processors/wait_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/checkpoint/processors/wait_test.py index 9f635b5c..f48fe6ce 100644 --- a/packages/aws-durable-execution-sdk-python-testing/tests/checkpoint/processors/wait_test.py +++ b/packages/aws-durable-execution-sdk-python-testing/tests/checkpoint/processors/wait_test.py @@ -59,7 +59,7 @@ def test_process_start_action(): wait_options=wait_options, ) - result = processor.process(update, None, notifier, execution_arn) + result = processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert isinstance(result, Operation) assert result.operation_id == "wait-123" @@ -88,7 +88,7 @@ def test_process_start_action_without_wait_options(): name="test-wait", ) - result = processor.process(update, None, notifier, execution_arn) + result = processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert isinstance(result, Operation) assert result.wait_details is not None @@ -111,7 +111,7 @@ def test_process_start_action_with_zero_seconds(): wait_options=wait_options, ) - result = processor.process(update, None, notifier, execution_arn) + result = processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert isinstance(result, Operation) assert result.wait_details is not None @@ -135,7 +135,7 @@ def test_process_start_action_with_parent_id(): wait_options=wait_options, ) - result = processor.process(update, None, notifier, execution_arn) + result = processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert result.parent_id == "parent-456" @@ -155,7 +155,7 @@ def test_process_start_action_with_sub_type(): wait_options=wait_options, ) - result = processor.process(update, None, notifier, execution_arn) + result = processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert result.sub_type == "timer" @@ -175,7 +175,9 @@ def test_process_cancel_action(): name="test-wait", ) - result = processor.process(update, current_op, notifier, execution_arn) + result = processor.process( + update, current_op, notifier, execution_arn, datetime.now(UTC) + ) assert isinstance(result, Operation) assert result.operation_id == "wait-123" @@ -195,7 +197,7 @@ def test_process_cancel_action_without_current_operation(): name="test-wait", ) - result = processor.process(update, None, notifier, execution_arn) + result = processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert isinstance(result, Operation) assert result.status == OperationStatus.CANCELLED @@ -216,7 +218,7 @@ def test_process_invalid_action(): with pytest.raises( InvalidParameterValueException, match="Invalid action for WAIT operation" ): - processor.process(update, None, notifier, execution_arn) + processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) def test_process_fail_action(): @@ -234,7 +236,7 @@ def test_process_fail_action(): with pytest.raises( InvalidParameterValueException, match="Invalid action for WAIT operation" ): - processor.process(update, None, notifier, execution_arn) + processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) def test_process_retry_action(): @@ -252,7 +254,7 @@ def test_process_retry_action(): with pytest.raises( InvalidParameterValueException, match="Invalid action for WAIT operation" ): - processor.process(update, None, notifier, execution_arn) + processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) def test_wait_details_created_correctly(): @@ -270,7 +272,7 @@ def test_wait_details_created_correctly(): ) before_time = datetime.now(UTC) - result = processor.process(update, None, notifier, execution_arn) + result = processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert result.wait_details.scheduled_end_timestamp > before_time @@ -289,7 +291,7 @@ def test_no_completed_or_failed_calls(): wait_options=wait_options, ) - processor.process(update, None, notifier, execution_arn) + processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert len(notifier.completed_calls) == 0 assert len(notifier.failed_calls) == 0 @@ -311,6 +313,6 @@ def test_cancel_no_timer_scheduled(): name="test-wait", ) - processor.process(update, current_op, notifier, execution_arn) + processor.process(update, current_op, notifier, execution_arn, datetime.now(UTC)) assert len(notifier.wait_timer_calls) == 0 diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/checkpoint/transformer_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/checkpoint/transformer_test.py index 3d433cdf..f152b0c5 100644 --- a/packages/aws-durable-execution-sdk-python-testing/tests/checkpoint/transformer_test.py +++ b/packages/aws-durable-execution-sdk-python-testing/tests/checkpoint/transformer_test.py @@ -8,6 +8,7 @@ from __future__ import annotations +from datetime import UTC, datetime from unittest.mock import Mock import pytest @@ -41,7 +42,7 @@ def __init__(self, return_value=None): self.return_value = return_value self.calls: list[tuple] = [] - def process(self, update, current_op, notifier, execution_arn): + def process(self, update, current_op, notifier, execution_arn, now=None): # noqa: ARG002 self.calls.append((update, current_op, notifier, execution_arn)) return self.return_value @@ -86,6 +87,7 @@ def test_apply_updates_with_empty_list_is_a_noop(): updates=[], client_token=None, touch=touched.append, + now=datetime.now(UTC), ) assert execution.operations == [] @@ -113,6 +115,7 @@ def test_apply_updates_unknown_type_raises(): updates=[update], client_token=None, touch=lambda _: None, + now=datetime.now(UTC), ) @@ -134,6 +137,7 @@ def test_apply_updates_skips_ops_when_processor_returns_none(): updates=[update], client_token=None, touch=touched.append, + now=datetime.now(UTC), ) assert execution.operations == [] @@ -162,6 +166,7 @@ def test_apply_updates_appends_new_operation_and_touches(): updates=[update], client_token=None, touch=touched.append, + now=datetime.now(UTC), ) assert execution.operations == [new_op] @@ -191,6 +196,7 @@ def test_apply_updates_replaces_existing_operation_in_place(): updates=[update], client_token=None, touch=lambda _: None, + now=datetime.now(UTC), ) assert execution.operations == [replaced] @@ -229,6 +235,7 @@ def test_apply_updates_preserves_order_across_multiple_updates(): ], client_token=None, touch=touched.append, + now=datetime.now(UTC), ) assert execution.operations == [op1, updated_op2, op3] @@ -245,6 +252,7 @@ def test_apply_updates_preserves_order_across_multiple_updates(): ], client_token=None, touch=touched.append, + now=datetime.now(UTC), ) assert execution.operations == [op1, updated_op2, op3, new_op4] assert touched == ["op2", "op4"] @@ -283,6 +291,7 @@ def test_apply_updates_dispatches_by_operation_type(): ], client_token=None, touch=lambda _: None, + now=datetime.now(UTC), ) assert execution.operations == [step_op, wait_op] @@ -310,6 +319,7 @@ def test_apply_updates_forwards_arn_notifier_and_current_op_to_processor(): updates=[update], client_token=None, touch=lambda _: None, + now=datetime.now(UTC), ) forwarded_update, forwarded_current_op, forwarded_notifier, forwarded_arn = ( @@ -339,6 +349,7 @@ def test_apply_updates_returns_completion_effect(): updates=[update], client_token=None, touch=lambda _: None, + now=datetime.now(UTC), ) assert effects == [ @@ -366,6 +377,7 @@ def test_apply_updates_returns_no_effects_for_plain_step(): ], client_token=None, touch=lambda _: None, + now=datetime.now(UTC), ) assert effects == [] @@ -393,6 +405,7 @@ def test_apply_updates_records_payload_size_for_paging(): updates=[update], client_token=None, touch=lambda _: None, + now=datetime.now(UTC), ) assert execution.operation_size_bytes["with-payload"] >= len(b"hello-world-payload") @@ -422,6 +435,7 @@ def test_apply_updates_records_size_for_error_payload(): updates=[update], client_token=None, touch=lambda _: None, + now=datetime.now(UTC), ) # Some non-zero size was recorded (exact value depends on @@ -451,6 +465,7 @@ def test_apply_updates_records_size_for_bytes_payload(): updates=[update], client_token=None, touch=lambda _: None, + now=datetime.now(UTC), ) # bytes payload length == 12. diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/clock_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/clock_test.py new file mode 100644 index 00000000..2ef890b2 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-testing/tests/clock_test.py @@ -0,0 +1,88 @@ +"""Tests for the durable timer clocks.""" + +from __future__ import annotations + +import datetime + +from aws_durable_execution_sdk_python_testing.clock import ( + RealClock, + SkipClock, + real_now, +) + + +def test_real_now_returns_utc() -> None: + now: datetime.datetime = real_now() + + assert now.tzinfo is datetime.UTC + + +def test_real_clock_arm_returns_remaining_wall_seconds() -> None: + clock = RealClock() + wake = clock.now() + datetime.timedelta(seconds=30) + + delay: float = clock.arm(wake) + + # The real clock waits the modeled duration, minus the tiny elapsed + # time between now() calls. + assert 29.0 <= delay <= 30.0 + + +def test_real_clock_arm_never_negative_for_past_wake() -> None: + clock = RealClock() + past = clock.now() - datetime.timedelta(seconds=30) + + assert clock.arm(past) == 0.0 + + +def test_skip_clock_arm_returns_zero_delay() -> None: + clock = SkipClock() + wake = clock.now() + datetime.timedelta(seconds=300) + + assert clock.arm(wake) == 0.0 + + +def test_skip_clock_now_reaches_armed_horizon() -> None: + clock = SkipClock() + wake = clock.now() + datetime.timedelta(seconds=300) + + clock.arm(wake) + + # Virtual now lands on the horizon, then keeps advancing with real + # time, so it never reads earlier than the armed wake. + assert wake <= clock.now() <= wake + datetime.timedelta(seconds=5) + + +def test_skip_clock_is_monotonic_across_arms() -> None: + clock = SkipClock() + start = clock.now() + far = start + datetime.timedelta(seconds=300) + near = start + datetime.timedelta(seconds=10) + + clock.arm(far) + clock.arm(near) + + # A later arm for an earlier horizon must not move virtual time back. + assert clock.now() >= far + + +def test_skip_clock_advances_with_real_time() -> None: + clock = SkipClock() + + first = clock.now() + second = clock.now() + + # Between arms the virtual clock tracks real time, so consecutive + # reads are ordered and distinct rather than frozen. + assert second >= first + + +def test_skip_clock_accumulates_skips() -> None: + clock = SkipClock() + start = clock.now() + + clock.arm(clock.now() + datetime.timedelta(seconds=100)) + clock.arm(clock.now() + datetime.timedelta(seconds=200)) + + elapsed = (clock.now() - start).total_seconds() + assert 300.0 <= elapsed <= 305.0 diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/execution_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/execution_test.py index f664e682..bbb2d0d1 100644 --- a/packages/aws-durable-execution-sdk-python-testing/tests/execution_test.py +++ b/packages/aws-durable-execution-sdk-python-testing/tests/execution_test.py @@ -78,11 +78,9 @@ def test_execution_new(mock_uuid4): assert execution.operations == [] -@patch("aws_durable_execution_sdk_python_testing.execution.datetime") -def test_execution_start(mock_datetime): +def test_execution_start(): """Test Execution.start method.""" mock_now = datetime(2023, 1, 1, 12, 0, 0, tzinfo=timezone.utc) - mock_datetime.now.return_value = mock_now start_input = StartDurableExecutionInput( account_id="123456789012", @@ -96,7 +94,7 @@ def test_execution_start(mock_datetime): ) execution = Execution("test-arn", start_input, []) - execution.start() + execution.start(now=mock_now) assert len(execution.operations) == 1 operation = execution.operations[0] @@ -481,11 +479,9 @@ def test_find_operation_not_exists(): execution.find_operation("non-existent-id") -@patch("aws_durable_execution_sdk_python_testing.execution.datetime") -def test_complete_wait_success(mock_datetime): +def test_complete_wait_success(): """Test complete_wait method successful completion.""" mock_now = datetime(2023, 1, 1, 12, 0, 0, tzinfo=timezone.utc) - mock_datetime.now.return_value = mock_now start_input = StartDurableExecutionInput( account_id="123456789012", @@ -506,7 +502,7 @@ def test_complete_wait_success(mock_datetime): ) execution = Execution("test-arn", start_input, [operation]) - result = execution.complete_wait("wait-op-id") + result = execution.complete_wait("wait-op-id", now=mock_now) assert result.status == OperationStatus.SUCCEEDED assert result.end_timestamp == mock_now diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/executor_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/executor_test.py index 0a4cae19..3c2738d0 100644 --- a/packages/aws-durable-execution-sdk-python-testing/tests/executor_test.py +++ b/packages/aws-durable-execution-sdk-python-testing/tests/executor_test.py @@ -2,7 +2,7 @@ import asyncio from datetime import UTC, datetime -from unittest.mock import Mock, patch +from unittest.mock import ANY, Mock, patch import pytest @@ -863,7 +863,7 @@ def test_should_complete_workflow_successfully_through_public_api( # Assert - Verify final execution status and stored results mock_store.load.assert_any_call(execution_arn="test-arn") - mock_execution.complete_success.assert_called_once_with(result="result") + mock_execution.complete_success.assert_called_once_with(result="result", now=ANY) mock_store.update.assert_called_once_with(mock_execution) mock_complete_events.assert_called_once_with(execution_arn="test-arn") @@ -883,7 +883,7 @@ def test_should_complete_workflow_with_failure_through_public_api( # Assert - Verify final execution status and stored error mock_store.load.assert_any_call(execution_arn="test-arn") - mock_execution.complete_fail.assert_called_once_with(error=error) + mock_execution.complete_fail.assert_called_once_with(error=error, now=ANY) mock_store.update.assert_called_once_with(mock_execution) mock_complete_events.assert_called_once_with(execution_arn="test-arn") @@ -902,7 +902,7 @@ def test_should_handle_workflow_completion_state_through_public_api( # Assert - Verify completion was processed and observer notifications sent mock_store.load.assert_any_call(execution_arn="test-arn") - mock_execution.complete_success.assert_called_once_with(result="result") + mock_execution.complete_success.assert_called_once_with(result="result", now=ANY) mock_store.update.assert_called_once_with(mock_execution) mock_complete_events.assert_called_once_with(execution_arn="test-arn") @@ -1252,7 +1252,7 @@ def test_complete_execution(executor, mock_store, mock_execution): executor.complete_execution("test-arn", "result") mock_store.load.assert_any_call(execution_arn="test-arn") - mock_execution.complete_success.assert_called_once_with(result="result") + mock_execution.complete_success.assert_called_once_with(result="result", now=ANY) mock_store.update.assert_called_once_with(mock_execution) mock_complete_events.assert_called_once_with(execution_arn="test-arn") @@ -1266,7 +1266,7 @@ def test_fail_execution(executor, mock_store, mock_execution): executor.fail_execution("test-arn", error) mock_store.load.assert_any_call(execution_arn="test-arn") - mock_execution.complete_fail.assert_called_once_with(error=error) + mock_execution.complete_fail.assert_called_once_with(error=error, now=ANY) mock_store.update.assert_called_once_with(mock_execution) mock_complete_events.assert_called_once_with(execution_arn="test-arn") @@ -2188,7 +2188,7 @@ def test_send_callback_success(executor, mock_store): assert isinstance(result, SendDurableExecutionCallbackSuccessResponse) mock_store.load.assert_any_call("test-arn") mock_execution.complete_callback_success.assert_called_once_with( - callback_id, b"success-result" + callback_id, b"success-result", now=ANY ) mock_store.update.assert_called_once_with(mock_execution) # Verify execution is invoked after callback success @@ -2226,7 +2226,7 @@ def test_send_callback_success_with_result(executor, mock_store): assert isinstance(result, SendDurableExecutionCallbackSuccessResponse) mock_execution.complete_callback_success.assert_called_once_with( - callback_id, b"test-result" + callback_id, b"test-result", now=ANY ) # Verify execution is invoked after callback success mock_invoke.assert_called_once_with("test-arn") @@ -2285,7 +2285,9 @@ def test_send_callback_failure_with_error(executor, mock_store): result = executor.send_callback_failure(callback_id, error) assert isinstance(result, SendDurableExecutionCallbackFailureResponse) - mock_execution.complete_callback_failure.assert_called_once_with(callback_id, error) + mock_execution.complete_callback_failure.assert_called_once_with( + callback_id, error, now=ANY + ) # Verify execution is invoked after callback failure mock_invoke.assert_called_once_with("test-arn") diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/skip_time_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/skip_time_test.py new file mode 100644 index 00000000..5d26e73c --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-testing/tests/skip_time_test.py @@ -0,0 +1,105 @@ +"""End-to-end tests for skip-time behaviour on the local runner.""" + +from __future__ import annotations + +from typing import Any + +from aws_durable_execution_sdk_python.config import Duration +from aws_durable_execution_sdk_python.context import DurableContext, durable_step +from aws_durable_execution_sdk_python.execution import ( + InvocationStatus, + durable_execution, +) +from aws_durable_execution_sdk_python.types import StepContext + +from aws_durable_execution_sdk_python_testing.runner import ( + DurableFunctionTestRunner, +) + + +_MODELED_WAIT_SECONDS = 300 + +_MODELED_MIDDLE_WAIT_SECONDS = 60 + + +@durable_execution +def _waits_then_returns(event: Any, context: DurableContext) -> str: # noqa: ARG001 + context.wait(Duration.from_seconds(_MODELED_WAIT_SECONDS)) + return "done" + + +def test_skip_time_completes_long_wait_and_keeps_faithful_history() -> None: + # skip_time defaults to True. A modeled 300s wait must complete well + # within the 30s execution timeout, which is only possible if the + # runner does not spend the modeled duration in wall-clock time. + with DurableFunctionTestRunner(handler=_waits_then_returns) as runner: + result = runner.run(input="x", execution_timeout=30) + + assert result.status is InvocationStatus.SUCCEEDED + + wait_ops = [op for op in result.operations if op.operation_type.value == "WAIT"] + assert len(wait_ops) == 1 + wait_op = wait_ops[0] + + assert wait_op.status.value == "SUCCEEDED" + assert wait_op.start_timestamp is not None + assert wait_op.scheduled_end_timestamp is not None + + # History records the real modeled duration, not the skipped + # wall-clock time. + modeled = ( + wait_op.scheduled_end_timestamp - wait_op.start_timestamp + ).total_seconds() + assert abs(modeled - _MODELED_WAIT_SECONDS) < 1.0 + + +@durable_step +def _step_a(step_context: StepContext) -> str: # noqa: ARG001 + return "a" + + +@durable_step +def _step_b(step_context: StepContext) -> str: # noqa: ARG001 + return "b" + + +@durable_execution +def _step_wait_step(event: Any, context: DurableContext) -> str: # noqa: ARG001 + context.step(_step_a()) + context.wait(Duration.from_seconds(_MODELED_MIDDLE_WAIT_SECONDS)) + context.step(_step_b()) + return "done" + + +def test_skip_time_keeps_history_monotonic_across_wait() -> None: + # stepA -> wait(60) -> stepB. Under skip, the clock is the single + # source of "now" for stamping, so the operation after the wait must + # start no earlier than the wait's modeled end. The history stays + # monotonic even though it is future-dated relative to wall-clock. + with DurableFunctionTestRunner(handler=_step_wait_step) as runner: + result = runner.run(input="x", execution_timeout=30) + + assert result.status is InvocationStatus.SUCCEEDED + + step_a = result.get_step("_step_a") + step_b = result.get_step("_step_b") + wait_ops = [op for op in result.operations if op.operation_type.value == "WAIT"] + assert len(wait_ops) == 1 + wait_op = wait_ops[0] + + assert wait_op.start_timestamp is not None + assert wait_op.scheduled_end_timestamp is not None + assert step_a.start_timestamp is not None + assert step_b.start_timestamp is not None + + # Faithful: the modeled wait duration is recorded exactly. + modeled = ( + wait_op.scheduled_end_timestamp - wait_op.start_timestamp + ).total_seconds() + assert abs(modeled - _MODELED_MIDDLE_WAIT_SECONDS) < 1.0 + + # Monotonic: the post-wait step cannot start before the wait's + # modeled end (the inversion the clock fix removes). + assert step_b.start_timestamp >= wait_op.scheduled_end_timestamp + # And the pre-wait step starts no later than the wait. + assert step_a.start_timestamp <= wait_op.start_timestamp