This commit is contained in:
@@ -2,97 +2,111 @@ import pytest
|
|||||||
from datetime import datetime, timedelta
|
from datetime import datetime, timedelta
|
||||||
from slic.utils.run_later import run_at, run_in, run_later, today, tomorrow, yesterday
|
from slic.utils.run_later import run_at, run_in, run_later, today, tomorrow, yesterday
|
||||||
|
|
||||||
# Simule tqdm
|
# Dummy progress bar replacement
|
||||||
class DummyBar:
|
class DummyBar:
|
||||||
def __enter__(self): return self
|
def __enter__(self): return self
|
||||||
def __exit__(self, *a): pass
|
def __exit__(self, *a): pass
|
||||||
def set(self, val): pass
|
def set(self, val): pass
|
||||||
|
|
||||||
# Simule l'heure courante
|
# Global clock control
|
||||||
class FakeClock:
|
class FakeDatetime:
|
||||||
now = datetime(2025, 1, 1, 6, 0, 0)
|
_now = None
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def set(cls, dt):
|
def set_now(cls, value):
|
||||||
cls.now = dt
|
cls._now = value
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def advance(cls, seconds):
|
def advance(cls, seconds):
|
||||||
cls.now += timedelta(seconds=seconds)
|
cls._now += timedelta(seconds=seconds)
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def now(cls):
|
||||||
|
return cls._now
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture(autouse=True)
|
@pytest.fixture
|
||||||
def patch_time(monkeypatch):
|
def patch_time(monkeypatch):
|
||||||
# Patch sleep + tqdm
|
# Mock sleep to do nothing
|
||||||
monkeypatch.setattr("slic.utils.run_later.sleep", lambda x: None)
|
monkeypatch.setattr("slic.utils.run_later.sleep", lambda x: None)
|
||||||
|
# Mock progress bar
|
||||||
monkeypatch.setattr("slic.utils.run_later.tqdm_mod", lambda *a, **k: DummyBar())
|
monkeypatch.setattr("slic.utils.run_later.tqdm_mod", lambda *a, **k: DummyBar())
|
||||||
|
# Mock datetime.now to use our fake time
|
||||||
# Patch datetime.now() dans slic.utils.run_later
|
monkeypatch.setattr("slic.utils.run_later.datetime", FakeDatetime)
|
||||||
class FakeDateTime(datetime):
|
# Mock helper functions to use fake time
|
||||||
@classmethod
|
monkeypatch.setattr("slic.utils.run_later.today", lambda *args, **kwargs: datetime(
|
||||||
def now(cls, tz=None):
|
FakeDatetime.now().year, FakeDatetime.now().month, FakeDatetime.now().day, *args, **kwargs))
|
||||||
return FakeClock.now
|
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.datetime", FakeDateTime)
|
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))
|
||||||
monkeypatch.setattr("slic.utils.run_later.today", lambda h=0, m=0: datetime(FakeClock.now.year, FakeClock.now.month, FakeClock.now.day, h, m))
|
|
||||||
monkeypatch.setattr("slic.utils.run_later.tomorrow", lambda h=0, m=0: datetime(FakeClock.now.year, FakeClock.now.month, FakeClock.now.day, h, m) + timedelta(days=1))
|
|
||||||
monkeypatch.setattr("slic.utils.run_later.yesterday", lambda h=0, m=0: datetime(FakeClock.now.year, FakeClock.now.month, FakeClock.now.day, h, m) - timedelta(days=1))
|
|
||||||
|
|
||||||
|
|
||||||
def test_run_at():
|
def test_run_at(patch_time):
|
||||||
triggered = []
|
triggered = []
|
||||||
def fexample(label): triggered.append(label)
|
def fexample(label): triggered.append(label)
|
||||||
|
|
||||||
FakeClock.set(datetime(2025, 1, 1, 6, 0, 0))
|
FakeDatetime.set_now(datetime(2025, 1, 1, 6, 0, 0))
|
||||||
run_at(today(6, 30), fexample, "not yet")
|
|
||||||
assert "not yet" not in triggered
|
|
||||||
|
|
||||||
FakeClock.advance(31 * 60)
|
# Schedule for 6:30
|
||||||
run_at(today(6, 30), fexample, "now OK")
|
run_at(today(6, 30), fexample, "now OK")
|
||||||
|
assert not triggered # Shouldn't have run yet
|
||||||
|
|
||||||
|
# Advance to 6:30:01
|
||||||
|
FakeDatetime.advance(30 * 60 + 1)
|
||||||
|
run_at(today(6, 30), fexample, "now OK") # This should trigger the function
|
||||||
assert "now OK" in triggered
|
assert "now OK" in triggered
|
||||||
|
|
||||||
FakeClock.set(datetime(2025, 1, 1, 6, 0, 0))
|
# Try scheduling something in the past
|
||||||
run_at(yesterday(23, 59), fexample, "too late")
|
run_at(yesterday(23, 59), fexample, "too late")
|
||||||
assert "too late" not in triggered
|
assert "too late" not in triggered
|
||||||
|
|
||||||
|
|
||||||
def test_run_in():
|
def test_run_in(patch_time):
|
||||||
triggered = []
|
triggered = []
|
||||||
def fexample(label): triggered.append(label)
|
def fexample(label): triggered.append(label)
|
||||||
|
|
||||||
FakeClock.set(datetime(2025, 1, 1, 12, 0, 0))
|
FakeDatetime.set_now(datetime(2025, 1, 1, 12, 0, 0))
|
||||||
run_in(timedelta(seconds=60), fexample, "not yet")
|
|
||||||
assert "not yet" not in triggered
|
|
||||||
|
|
||||||
FakeClock.advance(61)
|
# Schedule for 60 seconds later
|
||||||
run_in(timedelta(seconds=60), fexample, "now OK")
|
run_in(timedelta(seconds=60), fexample, "now OK")
|
||||||
|
assert not triggered # Shouldn't have run yet
|
||||||
|
|
||||||
|
# Advance 61 seconds
|
||||||
|
FakeDatetime.advance(61)
|
||||||
|
run_in(timedelta(seconds=60), fexample, "now OK") # This should trigger
|
||||||
assert "now OK" in triggered
|
assert "now OK" in triggered
|
||||||
|
|
||||||
FakeClock.advance(-200)
|
# Try scheduling something in the past
|
||||||
run_in(timedelta(seconds=-1), fexample, "too late")
|
run_in(timedelta(seconds=-1), fexample, "too late")
|
||||||
assert "too late" not in triggered
|
assert "too late" not in triggered
|
||||||
|
|
||||||
|
|
||||||
def test_run_later():
|
def test_run_later(patch_time):
|
||||||
triggered = []
|
triggered = []
|
||||||
def fexample(label): triggered.append(label)
|
def fexample(label): triggered.append(label)
|
||||||
|
|
||||||
FakeClock.set(datetime(2025, 1, 1, 14, 0, 0))
|
FakeDatetime.set_now(datetime(2025, 1, 1, 14, 0, 0))
|
||||||
run_later(45, fexample, "not yet")
|
|
||||||
assert "not yet" not in triggered
|
|
||||||
|
|
||||||
FakeClock.advance(60)
|
# Schedule for 45 seconds later
|
||||||
run_later(45, fexample, "now OK")
|
run_later(45, fexample, "now OK")
|
||||||
|
assert not triggered # Shouldn't have run yet
|
||||||
|
|
||||||
|
# Advance 60 seconds
|
||||||
|
FakeDatetime.advance(60)
|
||||||
|
run_later(45, fexample, "now OK") # This should trigger
|
||||||
assert "now OK" in triggered
|
assert "now OK" in triggered
|
||||||
|
|
||||||
|
# Try scheduling something in the past
|
||||||
run_later(timedelta(seconds=-1), fexample, "too late")
|
run_later(timedelta(seconds=-1), fexample, "too late")
|
||||||
assert "too late" not in triggered
|
assert "too late" not in triggered
|
||||||
|
|
||||||
FakeClock.set(datetime(2025, 1, 1, 15, 0, 0))
|
# Test with datetime object
|
||||||
run_later(today(15, 5), fexample, "not yet dt")
|
FakeDatetime.set_now(datetime(2025, 1, 1, 15, 0, 0))
|
||||||
assert "not yet dt" not in triggered
|
|
||||||
|
|
||||||
FakeClock.advance(6 * 60)
|
|
||||||
run_later(today(15, 5), fexample, "now OK dt")
|
run_later(today(15, 5), fexample, "now OK dt")
|
||||||
assert "now OK dt" in triggered
|
assert not triggered # Shouldn't have run yet
|
||||||
|
|
||||||
|
# Advance 6 minutes
|
||||||
|
FakeDatetime.advance(6 * 60)
|
||||||
|
run_later(today(15, 5), fexample, "now OK dt") # This should trigger
|
||||||
|
assert "now OK dt" in triggered
|
||||||
Reference in New Issue
Block a user