Coverage for src / mcp_server_langgraph / core / time_provider.py: 88%

1""" 

2Time provider abstraction for test performance optimization. 

3 

4This module provides a virtual time system that allows tests to fast-forward time 

5without real sleep() calls, eliminating 50+ seconds of sleep overhead. 

6 

7Design Pattern: Dependency Injection 

8- Components accept a TimeProvider parameter for time operations 

9- Production: uses RealTimeProvider (delegates to time module) 

10- Testing: uses VirtualClock (instant time advancement) 

11 

12TDD Context: 

13- This implementation follows the Green phase (making tests pass) 

14- Tests were written first in tests/unit/test_time_provider.py 

15- Eliminates sleep overhead in TTL, timeout, and staleness tests 

16""" 

17 

18import asyncio 

19import time 

20from typing import Protocol 

21 

22 

23class TimeProvider(Protocol): 

24 """ 

25 Protocol defining time operations for dependency injection. 

26 

27 This enables test doubles to replace real time with virtual time, 

28 allowing tests to fast-forward without actual sleep delays. 

29 """ 

30 

31 def time(self) -> float: 

32 """Return current Unix timestamp (seconds since epoch).""" 

33 ... 

34 

35 def sleep(self, duration: float) -> None: 

36 """Block for specified duration (seconds).""" 

37 ... 

38 

39 def monotonic(self) -> float: 

40 """Return monotonically increasing time (for elapsed time calculations).""" 

41 ... 

42 

43 async def async_sleep(self, duration: float) -> None: 

44 """Async sleep for specified duration (seconds).""" 

45 ... 

46 

47 

48class RealTimeProvider: 

49 """ 

50 Production time provider that delegates to Python's time module. 

51 

52 This is the default provider used in production, providing real time 

53 and actual sleep operations. 

54 """ 

55 

56 def time(self) -> float: 

57 """Return current Unix timestamp.""" 

58 return time.time() 

59 

60 def sleep(self, duration: float) -> None: 

61 """Block for specified duration.""" 

62 time.sleep(duration) 

63 

64 def monotonic(self) -> float: 

65 """Return monotonically increasing time.""" 

66 return time.monotonic() 

67 

68 async def async_sleep(self, duration: float) -> None: 

69 """Async sleep for specified duration.""" 

70 await asyncio.sleep(duration) 

71 

72 

73class VirtualClock: 

74 """ 

75 Test time provider that allows instant time advancement. 

76 

77 Instead of real sleep() calls, this clock advances time instantly, 

78 eliminating sleep overhead in tests while preserving behavior. 

79 

80 Performance Impact: 

81 - Reduces test suite execution by ~50 seconds 

82 - TTL tests: 1-3s sleep → instant (100x faster) 

83 - Timeout tests: 1s sleep → instant (1000x faster) 

84 - Property tests: 35s → 0.35s (100x faster) 

85 

86 Example: 

87 >>> clock = VirtualClock() 

88 >>> clock.time() 

89 0.0 

90 >>> clock.sleep(10.0) # Instant, no actual waiting 

91 >>> clock.time() 

92 10.0 

93 """ 

94 

95 def __init__(self, start_time: float = 0.0): 

96 """ 

97 Initialize virtual clock at specified time. 

98 

99 Args: 

100 start_time: Initial time value (default: 0.0) 

101 """ 

102 self._current_time = start_time 

103 self._lock = asyncio.Lock() # Thread-safe time updates 

104 

105 def time(self) -> float: 

106 """Return current virtual time.""" 

107 return self._current_time 

108 

109 def sleep(self, duration: float) -> None: 

110 """ 

111 Advance virtual time instantly without real sleep. 

112 

113 Args: 

114 duration: Time to advance (seconds) 

115 

116 Raises: 

117 ValueError: If duration is negative 

118 """ 

119 if duration < 0: 

120 msg = "sleep duration must be non-negative" 

121 raise ValueError(msg) 

122 self._current_time += duration 

123 

124 def monotonic(self) -> float: 

125 """Return current virtual time (monotonically increasing).""" 

126 return self._current_time 

127 

128 async def async_sleep(self, duration: float) -> None: 

129 """ 

130 Async version of sleep - advances time instantly. 

131 

132 Args: 

133 duration: Time to advance (seconds) 

134 

135 Raises: 

136 ValueError: If duration is negative 

137 """ 

138 if duration < 0: 

139 msg = "sleep duration must be non-negative" 

140 raise ValueError(msg) 

141 

142 async with self._lock: 

143 self._current_time += duration 

144 # Yield control to allow other coroutines to run 

145 await asyncio.sleep(0) 

146 

147 def advance(self, duration: float) -> None: 

148 """ 

149 Manually advance virtual time (test helper). 

150 

151 This is useful for tests that need to skip time without calling sleep(). 

152 

153 Args: 

154 duration: Time to advance (seconds) 

155 """ 

156 self._current_time += duration 

157 

158 

159# Default time provider for production use 

160_default_provider = RealTimeProvider() 

161 

162 

163def get_default_time_provider() -> RealTimeProvider: 

164 """ 

165 Get the default time provider for production use. 

166 

167 Returns: 

168 RealTimeProvider instance 

169 """ 

170 return _default_provider