This commit is contained in:
@@ -2,102 +2,106 @@ import pytest
|
||||
from datetime import datetime, timedelta
|
||||
from slic.utils.run_later import run_at, run_in, run_later, today, tomorrow, yesterday
|
||||
|
||||
# Dummy tqdm
|
||||
# Dummy progress bar replacement
|
||||
class DummyBar:
|
||||
def __enter__(self): return self
|
||||
def __exit__(self, *a): pass
|
||||
def set(self, val): pass
|
||||
|
||||
# Fake time controller
|
||||
# Global clock control
|
||||
class FakeDatetime:
|
||||
_now = None
|
||||
|
||||
@classmethod
|
||||
def set_now(cls, dt):
|
||||
cls._now = dt
|
||||
|
||||
@classmethod
|
||||
def now(cls):
|
||||
return cls._now
|
||||
def set_now(cls, value):
|
||||
cls._now = value
|
||||
|
||||
@classmethod
|
||||
def advance(cls, seconds):
|
||||
cls._now += timedelta(seconds=seconds)
|
||||
|
||||
# Wrapper to patch the datetime module
|
||||
class FakeDateTimeModule:
|
||||
def __getattr__(self, attr):
|
||||
if attr == "now":
|
||||
return FakeDatetime.now
|
||||
return getattr(datetime, attr)
|
||||
@classmethod
|
||||
def now(cls):
|
||||
return cls._now
|
||||
|
||||
# -------- Tests -------- #
|
||||
|
||||
def test_run_at(monkeypatch):
|
||||
@pytest.fixture
|
||||
def patch_time(monkeypatch):
|
||||
monkeypatch.setattr("slic.utils.run_later.sleep", lambda x: None)
|
||||
monkeypatch.setattr("slic.utils.run_later.tqdm_mod", lambda *a, **k: DummyBar())
|
||||
monkeypatch.setattr("slic.utils.run_later.today", lambda *args, **kwargs: datetime(
|
||||
FakeDatetime.now().year, FakeDatetime.now().month, FakeDatetime.now().day, *args, **kwargs))
|
||||
monkeypatch.setattr("slic.utils.run_later.tomorrow", lambda *args, **kwargs: datetime(
|
||||
FakeDatetime.now().year, FakeDatetime.now().month, FakeDatetime.now().day, *args, **kwargs) + timedelta(days=1))
|
||||
monkeypatch.setattr("slic.utils.run_later.yesterday", lambda *args, **kwargs: datetime(
|
||||
FakeDatetime.now().year, FakeDatetime.now().month, FakeDatetime.now().day, *args, **kwargs) - timedelta(days=1))
|
||||
|
||||
|
||||
def test_run_at(patch_time):
|
||||
triggered = []
|
||||
def fexample(label): triggered.append(label)
|
||||
|
||||
FakeDatetime.set_now(datetime(2025, 1, 1, 6, 0, 0))
|
||||
monkeypatch.setattr("slic.utils.run_later.datetime", FakeDateTimeModule())
|
||||
monkeypatch.setattr("slic.utils.run_later.sleep", lambda x: None)
|
||||
monkeypatch.setattr("slic.utils.run_later.tqdm_mod", lambda *a, **k: DummyBar())
|
||||
|
||||
# Trop tôt : ne doit pas exécuter
|
||||
# Trop tôt
|
||||
run_at(today(6, 30), fexample, "not yet")
|
||||
assert "not yet" not in triggered
|
||||
|
||||
# Bon moment : doit exécuter
|
||||
FakeDatetime.advance(31 * 60) # +31 min
|
||||
# Moment déclenché
|
||||
FakeDatetime.advance(31 * 60)
|
||||
run_at(today(6, 30), fexample, "now OK")
|
||||
assert "now OK" in triggered
|
||||
|
||||
# Trop tard : ne doit pas exécuter
|
||||
# Trop tard
|
||||
run_at(yesterday(23, 59), fexample, "too late")
|
||||
assert "too late" not in triggered
|
||||
|
||||
|
||||
def test_run_in(monkeypatch):
|
||||
triggered = []
|
||||
def fexample(label): triggered.append(label)
|
||||
|
||||
FakeDatetime.set_now(datetime(2025, 1, 1, 10, 0, 0))
|
||||
monkeypatch.setattr("slic.utils.run_later.datetime", FakeDateTimeModule())
|
||||
monkeypatch.setattr("slic.utils.run_later.sleep", lambda x: None)
|
||||
monkeypatch.setattr("slic.utils.run_later.tqdm_mod", lambda *a, **k: DummyBar())
|
||||
|
||||
# Trop tôt
|
||||
run_in(timedelta(minutes=10), fexample, "not yet")
|
||||
assert "not yet" not in triggered
|
||||
|
||||
# Bon moment
|
||||
FakeDatetime.advance(11 * 60)
|
||||
run_in(timedelta(minutes=10), fexample, "now OK")
|
||||
assert "now OK" in triggered
|
||||
|
||||
# Trop tard
|
||||
FakeDatetime.advance(-3600)
|
||||
run_in(timedelta(seconds=-1), fexample, "too late")
|
||||
assert "too late" not in triggered
|
||||
|
||||
|
||||
def test_run_later(monkeypatch):
|
||||
def test_run_in(patch_time):
|
||||
triggered = []
|
||||
def fexample(label): triggered.append(label)
|
||||
|
||||
FakeDatetime.set_now(datetime(2025, 1, 1, 12, 0, 0))
|
||||
monkeypatch.setattr("slic.utils.run_later.datetime", FakeDateTimeModule())
|
||||
monkeypatch.setattr("slic.utils.run_later.sleep", lambda x: None)
|
||||
monkeypatch.setattr("slic.utils.run_later.tqdm_mod", lambda *a, **k: DummyBar())
|
||||
|
||||
# Trop tôt (15 secondes à attendre)
|
||||
run_later(15, fexample, "not yet")
|
||||
# Trop tôt
|
||||
run_in(timedelta(seconds=60), fexample, "not yet")
|
||||
assert "not yet" not in triggered
|
||||
|
||||
# Bon moment (20 secondes plus tard)
|
||||
FakeDatetime.advance(20)
|
||||
run_later(15, fexample, "now OK")
|
||||
# Moment déclenché
|
||||
FakeDatetime.advance(61)
|
||||
run_in(timedelta(seconds=60), fexample, "now OK")
|
||||
assert "now OK" in triggered
|
||||
|
||||
# Trop tard (temps déjà dépassé)
|
||||
run_later(timedelta(seconds=-5), fexample, "too late")
|
||||
# Trop tard
|
||||
FakeDatetime.advance(-100)
|
||||
run_in(timedelta(seconds=-1), fexample, "too late")
|
||||
assert "too late" not in triggered
|
||||
|
||||
|
||||
def test_run_later(patch_time):
|
||||
triggered = []
|
||||
def fexample(label): triggered.append(label)
|
||||
|
||||
FakeDatetime.set_now(datetime(2025, 1, 1, 14, 0, 0))
|
||||
|
||||
# Trop tôt (float seconds)
|
||||
run_later(45, fexample, "not yet")
|
||||
assert "not yet" not in triggered
|
||||
|
||||
# Moment déclenché
|
||||
FakeDatetime.advance(60)
|
||||
run_later(45, fexample, "now OK")
|
||||
assert "now OK" in triggered
|
||||
|
||||
# Trop tard
|
||||
run_later(timedelta(seconds=-1), fexample, "too late")
|
||||
assert "too late" not in triggered
|
||||
|
||||
# Format datetime
|
||||
FakeDatetime.set_now(datetime(2025, 1, 1, 15, 0, 0))
|
||||
run_later(today(15, 5), fexample, "not yet dt")
|
||||
assert "not yet dt" not in triggered
|
||||
|
||||
FakeDatetime.advance(6 * 60)
|
||||
run_later(today(15, 5), fexample, "now OK dt")
|
||||
assert "now OK dt" in triggered
|
||||
|
||||
Reference in New Issue
Block a user