This commit is contained in:
@@ -1,65 +1,86 @@
|
||||
import pytest
|
||||
from datetime import datetime, timedelta
|
||||
from slic.utils.run_later import run_at, run_in, run_later, today, tomorrow, yesterday
|
||||
from freezegun import freeze_time
|
||||
|
||||
class DummyBar:
|
||||
def __enter__(self): return self
|
||||
def __exit__(self, *a): pass
|
||||
def set(self, val): pass
|
||||
|
||||
class FakeDatetime:
|
||||
_now = None
|
||||
|
||||
@classmethod
|
||||
def set_now(cls, value):
|
||||
cls._now = value
|
||||
|
||||
@classmethod
|
||||
def advance(cls, seconds):
|
||||
cls._now += timedelta(seconds=seconds)
|
||||
|
||||
@classmethod
|
||||
def now(cls):
|
||||
return cls._now
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def patch_helpers(monkeypatch):
|
||||
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.datetime.now", lambda: FakeDatetime.now())
|
||||
|
||||
@freeze_time("2025-01-01 06:00:00")
|
||||
def test_run_at():
|
||||
triggered = []
|
||||
def fexample(label): triggered.append(label)
|
||||
|
||||
# Trop tôt
|
||||
FakeDatetime.set_now(datetime(2025, 1, 1, 6, 0))
|
||||
|
||||
run_at(today(6, 30), fexample, "not yet")
|
||||
assert "not yet" not in triggered
|
||||
|
||||
@freeze_time("2025-01-01 06:31:00")
|
||||
def test_run_at_triggered():
|
||||
triggered = []
|
||||
def fexample(label): triggered.append(label)
|
||||
|
||||
FakeDatetime.advance(31 * 60)
|
||||
run_at(today(6, 30), fexample, "now OK")
|
||||
assert "now OK" in triggered
|
||||
|
||||
@freeze_time("2025-01-01 12:00:00")
|
||||
def test_run_in_and_later():
|
||||
run_at(yesterday(23, 59), fexample, "too late")
|
||||
assert "too late" not in triggered
|
||||
|
||||
def test_run_in():
|
||||
triggered = []
|
||||
def fexample(label): triggered.append(label)
|
||||
|
||||
# Avant le moment
|
||||
FakeDatetime.set_now(datetime(2025, 1, 1, 12, 0))
|
||||
|
||||
run_in(timedelta(seconds=60), fexample, "not yet")
|
||||
assert "not yet" not in triggered
|
||||
|
||||
@freeze_time("2025-01-01 12:01:01")
|
||||
def test_run_in_triggered():
|
||||
triggered = []
|
||||
def fexample(label): triggered.append(label)
|
||||
|
||||
FakeDatetime.advance(61)
|
||||
run_in(timedelta(seconds=60), fexample, "now OK")
|
||||
assert "now OK" in triggered
|
||||
|
||||
@freeze_time("2025-01-01 15:00:00")
|
||||
def test_run_later_datetime():
|
||||
FakeDatetime.advance(-100)
|
||||
run_in(timedelta(seconds=-1), fexample, "too late")
|
||||
assert "too late" not in triggered
|
||||
|
||||
def test_run_later():
|
||||
triggered = []
|
||||
def fexample(label): triggered.append(label)
|
||||
|
||||
# Pas encore
|
||||
run_later(today(15, 5), fexample, "not yet")
|
||||
FakeDatetime.set_now(datetime(2025, 1, 1, 14, 0))
|
||||
|
||||
run_later(45, fexample, "not yet")
|
||||
assert "not yet" not in triggered
|
||||
|
||||
@freeze_time("2025-01-01 15:06:00")
|
||||
def test_run_later_datetime_triggered():
|
||||
triggered = []
|
||||
def fexample(label): triggered.append(label)
|
||||
|
||||
run_later(today(15, 5), fexample, "now OK")
|
||||
FakeDatetime.advance(60)
|
||||
run_later(45, fexample, "now OK")
|
||||
assert "now OK" in triggered
|
||||
|
||||
run_later(timedelta(seconds=-1), fexample, "too late")
|
||||
assert "too late" not in triggered
|
||||
|
||||
FakeDatetime.set_now(datetime(2025, 1, 1, 15, 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