The virtual-time loop driving the current call, or None. Returns None when there is no running loop (e.g. a background thread such as the OpenTelemetry metrics exporter) or the running loop is an ordinary real one. The clock patch lives on the `time` module process-wide for the duration
()
| 58 | |
| 59 | |
| 60 | def _virtual_loop() -> _VirtualTimeLoop | None: |
| 61 | """The virtual-time loop driving the current call, or None. |
| 62 | |
| 63 | Returns None when there is no running loop (e.g. a background thread such as the OpenTelemetry |
| 64 | metrics exporter) or the running loop is an ordinary real one. The clock patch lives on the |
| 65 | `time` module process-wide for the duration of a virtual-time test, so it is reachable from |
| 66 | any thread; only callers actually on the virtual loop should see virtual time -- everyone else |
| 67 | must get the real wall clock, otherwise a stray read freezes at the epoch (which, for the |
| 68 | metrics exporter, stamps batches with a 2023 timestamp the gateway then rejects). |
| 69 | """ |
| 70 | try: |
| 71 | loop = asyncio.get_running_loop() |
| 72 | except RuntimeError: |
| 73 | return None |
| 74 | return loop if isinstance(loop, _VirtualTimeLoop) else None |
| 75 | |
| 76 | |
| 77 | def _now() -> float: |