test: the app's clock never goes back in the test suite
The docker VM the suite runs in steps its wall clock back now and then (measured: 2 steps in 20 s, up to 0.5 s), so rows a test made one after another could read out of order (a files list sorted newest first, a scheduler's last runs). Every app module that took datetime from the datetime module reads a clock that never repeats or goes back; the scheduler test's own fixture for this goes.
This commit is contained in:
@@ -13,6 +13,7 @@ that the migration chain applies cleanly. Each test then gets a clean database v
|
||||
tests that COMMIT (the whole point of this lane — transaction-ownership behaviour) don't leak.
|
||||
"""
|
||||
|
||||
import datetime as _dt
|
||||
import os
|
||||
import shutil
|
||||
from urllib.parse import quote
|
||||
@@ -317,6 +318,41 @@ def _reset_rate_limiters():
|
||||
obj.reset()
|
||||
|
||||
|
||||
class _NeverBack(_dt.datetime):
|
||||
"""The app's clock in tests: the wall clock, but never repeating and never going back — each
|
||||
reading at least a microsecond after the last one in this process."""
|
||||
|
||||
last: _dt.datetime | None = None
|
||||
clock: type[_dt.datetime] = (
|
||||
_dt.datetime
|
||||
) # what it reads; a probe can put a stepping one under it
|
||||
|
||||
@classmethod
|
||||
def now(cls, tz=None):
|
||||
t = cls.clock.now(tz)
|
||||
last = cls.last
|
||||
if last is not None and (last.tzinfo is None) == (t.tzinfo is None) and t <= last:
|
||||
t = last + _dt.timedelta(microseconds=1)
|
||||
cls.last = t
|
||||
return t
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _clock_never_goes_back(monkeypatch):
|
||||
"""Rows a test makes one after another are stamped in that order. The wall clock of the docker
|
||||
VM the suite runs in steps BACK now and then (measured: 2 steps in 20 s, up to 0.5 s), and a
|
||||
test that makes three files and reads them newest first then got them out of order. Every loaded
|
||||
`app.*` module that took `datetime` from the datetime module reads this clock instead (a sweep,
|
||||
as the limiter reset below does, so a new module is covered the day it lands). A database
|
||||
default (`server_default=func.now()`) is Postgres's clock and is not covered: a test that orders
|
||||
such rows made within a second of each other gives them distinct times itself."""
|
||||
import sys
|
||||
|
||||
for name, mod in list(sys.modules.items()):
|
||||
if (name == "app" or name.startswith("app.")) and vars(mod).get("datetime") is _dt.datetime:
|
||||
monkeypatch.setattr(mod, "datetime", _NeverBack)
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _reset_log_erasures():
|
||||
"""`logs._erased` is module-level and remembers an erased account for an hour of monotonic time
|
||||
|
||||
Reference in New Issue
Block a user