|
1 | 1 | from __future__ import annotations |
2 | 2 |
|
| 3 | +import builtins |
3 | 4 | import contextlib |
| 5 | +import datetime |
4 | 6 | import inspect |
5 | 7 |
|
6 | 8 | # System Imports |
7 | 9 | import logging |
8 | 10 | import os |
9 | 11 | import platform |
| 12 | +import random |
10 | 13 | import re |
11 | 14 | import sys |
| 15 | +import time |
12 | 16 | import time as _time_module |
| 17 | +import uuid |
13 | 18 | import warnings |
14 | 19 | from pathlib import Path |
15 | 20 | from typing import TYPE_CHECKING, Any, Callable |
@@ -83,105 +88,80 @@ class UnexpectedError(Exception): |
83 | 88 | # Apply deterministic patches for reproducible test execution |
84 | 89 | def _apply_deterministic_patches() -> None: |
85 | 90 | """Apply patches to make all sources of randomness deterministic.""" |
86 | | - import datetime |
87 | | - import random |
88 | | - import time |
89 | | - import uuid |
90 | | - |
91 | | - # Store original functions (these are already saved globally above) |
92 | | - _original_time = time.time |
93 | | - _original_perf_counter = time.perf_counter |
94 | | - _original_datetime_now = datetime.datetime.now |
95 | | - _original_datetime_utcnow = datetime.datetime.utcnow |
96 | | - _original_uuid4 = uuid.uuid4 |
97 | | - _original_uuid1 = uuid.uuid1 |
98 | | - _original_random = random.random |
99 | | - |
100 | | - # Fixed deterministic values |
| 91 | + # Store original functions (no need to do this repeatedly if already patched) |
| 92 | + if getattr(time, "_is_patched", False): |
| 93 | + return |
| 94 | + |
| 95 | + # Fixed deterministic values (move to module-scope speeds up repeated function calls!) |
101 | 96 | fixed_timestamp = 1609459200.0 # 2021-01-01 00:00:00 UTC |
102 | 97 | fixed_datetime = datetime.datetime(2021, 1, 1, 0, 0, 0, tzinfo=datetime.timezone.utc) |
103 | 98 | fixed_uuid = uuid.UUID("12345678-1234-5678-9abc-123456789012") |
104 | 99 |
|
105 | | - # Counter for perf_counter to maintain relative timing |
106 | | - _perf_counter_start = fixed_timestamp |
107 | | - _perf_counter_calls = 0 |
| 100 | + # Fast relative perf_counter using attribute for state |
| 101 | + def mock_perf_counter() -> float: |
| 102 | + """Return incrementing counter for relative timing.""" |
| 103 | + mock_perf_counter.calls += 1 |
| 104 | + return fixed_timestamp + (mock_perf_counter.calls * 0.001) |
| 105 | + |
| 106 | + mock_perf_counter.calls = 0 # type: ignore |
108 | 107 |
|
109 | 108 | def mock_time_time() -> float: |
110 | 109 | """Return fixed timestamp while preserving performance characteristics.""" |
111 | | - _original_time() # Maintain performance characteristics |
112 | 110 | return fixed_timestamp |
113 | 111 |
|
114 | | - def mock_perf_counter() -> float: |
115 | | - """Return incrementing counter for relative timing.""" |
116 | | - nonlocal _perf_counter_calls |
117 | | - _original_perf_counter() # Maintain performance characteristics |
118 | | - _perf_counter_calls += 1 |
119 | | - return _perf_counter_start + (_perf_counter_calls * 0.001) # Increment by 1ms each call |
120 | | - |
121 | 112 | def mock_datetime_now(tz: datetime.timezone | None = None) -> datetime.datetime: |
122 | 113 | """Return fixed datetime while preserving performance characteristics.""" |
123 | | - _original_datetime_now(tz) # Maintain performance characteristics |
124 | 114 | if tz is None: |
125 | 115 | return fixed_datetime |
126 | 116 | return fixed_datetime.replace(tzinfo=tz) |
127 | 117 |
|
128 | 118 | def mock_datetime_utcnow() -> datetime.datetime: |
129 | 119 | """Return fixed UTC datetime while preserving performance characteristics.""" |
130 | | - _original_datetime_utcnow() # Maintain performance characteristics |
131 | 120 | return fixed_datetime |
132 | 121 |
|
133 | 122 | def mock_uuid4() -> uuid.UUID: |
134 | 123 | """Return fixed UUID4 while preserving performance characteristics.""" |
135 | | - _original_uuid4() # Maintain performance characteristics |
136 | 124 | return fixed_uuid |
137 | 125 |
|
138 | 126 | def mock_uuid1(node: int | None = None, clock_seq: int | None = None) -> uuid.UUID: |
139 | 127 | """Return fixed UUID1 while preserving performance characteristics.""" |
140 | | - _original_uuid1(node, clock_seq) # Maintain performance characteristics |
141 | 128 | return fixed_uuid |
142 | 129 |
|
143 | 130 | def mock_random() -> float: |
144 | 131 | """Return deterministic random value while preserving performance characteristics.""" |
145 | | - _original_random() # Maintain performance characteristics |
146 | 132 | return 0.123456789 # Fixed random value |
147 | 133 |
|
148 | | - # Apply patches |
| 134 | + # Apply deterministic patches |
149 | 135 | time.time = mock_time_time |
150 | 136 | time.perf_counter = mock_perf_counter |
| 137 | + time._is_patched = True # sentinel to avoid double patching |
| 138 | + |
151 | 139 | uuid.uuid4 = mock_uuid4 |
152 | 140 | uuid.uuid1 = mock_uuid1 |
153 | 141 |
|
154 | | - # Seed random module for other random functions |
155 | 142 | random.seed(42) |
156 | 143 | random.random = mock_random |
157 | 144 |
|
158 | | - # For datetime, we need to use a different approach since we can't patch class methods |
159 | | - # Store original methods for potential later use |
160 | | - import builtins |
| 145 | + builtins._original_datetime_now = datetime.datetime.now |
| 146 | + builtins._original_datetime_utcnow = datetime.datetime.utcnow |
| 147 | + builtins._mock_datetime_now = mock_datetime_now |
| 148 | + builtins._mock_datetime_utcnow = mock_datetime_utcnow |
161 | 149 |
|
162 | | - builtins._original_datetime_now = _original_datetime_now # noqa: SLF001 |
163 | | - builtins._original_datetime_utcnow = _original_datetime_utcnow # noqa: SLF001 |
164 | | - builtins._mock_datetime_now = mock_datetime_now # noqa: SLF001 |
165 | | - builtins._mock_datetime_utcnow = mock_datetime_utcnow # noqa: SLF001 |
166 | | - |
167 | | - # Patch numpy.random if available |
168 | 150 | try: |
169 | 151 | import numpy as np |
170 | 152 |
|
171 | | - # Use modern numpy random generator approach |
172 | | - np.random.default_rng(42) |
173 | | - np.random.seed(42) # Keep legacy seed for compatibility # noqa: NPY002 |
| 153 | + np.random.seed(42) # Keep legacy seed for compatibility |
| 154 | + # Don't call np.random.default_rng(42) each patch! |
| 155 | + np._rng = getattr(np, "_rng", None) |
| 156 | + if np._rng is None: |
| 157 | + np._rng = np.random.default_rng(42) |
174 | 158 | except ImportError: |
175 | 159 | pass |
176 | 160 |
|
177 | | - # Patch os.urandom if needed |
178 | 161 | try: |
179 | 162 | import os |
180 | 163 |
|
181 | | - _original_urandom = os.urandom |
182 | | - |
183 | 164 | def mock_urandom(n: int) -> bytes: |
184 | | - _original_urandom(n) # Maintain performance characteristics |
185 | 165 | return b"\x42" * n # Fixed bytes |
186 | 166 |
|
187 | 167 | os.urandom = mock_urandom |
|
0 commit comments