Coverage for async-durable-execution/src/async_durable_execution/runner/local/time_scale.py: 76%

1"""Helpers for scaling local durable timer delays in tests.""" 

2 

3from __future__ import annotations 

4 

5import logging 

6import os 

7 

8logger = logging.getLogger(__name__) 

9 

10_TIME_SCALE_ENV = "DURABLE_EXECUTION_TIME_SCALE" 

11 

12 

13def get_time_scale() -> float: 

14 """Return the configured local durable timer scale.""" 

15 raw_scale = os.getenv(_TIME_SCALE_ENV, "1.0") 

16 try: 

17 scale = float(raw_scale) 

18 except ValueError: 

19 logger.warning("Ignoring invalid %s value: %s", _TIME_SCALE_ENV, raw_scale) 

20 return 1.0 

21 

22 if scale < 0: 

23 logger.warning("Ignoring negative %s value: %s", _TIME_SCALE_ENV, raw_scale) 

24 return 1.0 

25 

26 return scale 

27 

28 

29def scale_delay(delay: float | int, *, minimum: float = 0) -> float: 

30 """Scale a durable timer delay for local testing.""" 

31 scaled_delay = float(delay) * get_time_scale() 

32 if minimum <= 0: 

33 return scaled_delay 

34 

35 return max(scaled_delay, min(float(delay), minimum))