126 lines
3.9 KiB
Python
126 lines
3.9 KiB
Python
import pytest
|
|
from slic.utils.run_later import run_at, run_in, run_later
|
|
from datetime import datetime, timedelta
|
|
|
|
# Dummy progress bar replacement to avoid actual tqdm usage
|
|
class DummyBar:
|
|
def __enter__(self): return self
|
|
def __exit__(self, *a): pass
|
|
def set(self, val): pass
|
|
|
|
|
|
# Helper class to simulate and manipulate time for testing
|
|
class FakeDatetime:
|
|
def __init__(self, start_time):
|
|
self.t = start_time
|
|
def now(self):
|
|
return self.t
|
|
def advance(self, seconds):
|
|
self.t += timedelta(seconds=seconds)
|
|
def __getattr__(self, attr):
|
|
return getattr(datetime, attr)
|
|
|
|
|
|
def test_run_at(monkeypatch, capsys):
|
|
triggered = []
|
|
|
|
def fexample(label):
|
|
print(f"run_at triggered: {label}")
|
|
triggered.append(label)
|
|
|
|
base = datetime(2025, 1, 1, 12, 0, 0)
|
|
fake_dt = FakeDatetime(base)
|
|
|
|
# Patch dependencies to control time and disable delays
|
|
monkeypatch.setattr("slic.utils.time_runner.datetime", fake_dt)
|
|
monkeypatch.setattr("slic.utils.time_runner.sleep", lambda x: None)
|
|
monkeypatch.setattr("slic.utils.time_runner.tqdm_mod", lambda *a, **k: DummyBar())
|
|
|
|
# Should NOT trigger because the target time is in the future
|
|
run_at(base + timedelta(seconds=30), fexample, "not yet")
|
|
assert "not yet" not in triggered
|
|
|
|
# Simulate time passing beyond the target : should trigger
|
|
fake_dt.advance(31)
|
|
run_at(base + timedelta(seconds=30), fexample, "now OK 0")
|
|
assert "now OK 0" in triggered
|
|
|
|
# Target time is in the past : should do nothing
|
|
run_at(base - timedelta(seconds=1), fexample, "too late")
|
|
assert "too late" not in triggered
|
|
|
|
out = capsys.readouterr().out
|
|
print(out)
|
|
|
|
|
|
def test_run_in(monkeypatch, capsys):
|
|
triggered = []
|
|
|
|
def fexample(label):
|
|
print(f"run_in triggered: {label}")
|
|
triggered.append(label)
|
|
|
|
base = datetime(2025, 1, 1, 13, 0, 0)
|
|
fake_dt = FakeDatetime(base)
|
|
|
|
# Patch time and delay mechanisms
|
|
monkeypatch.setattr("slic.utils.time_runner.datetime", fake_dt)
|
|
monkeypatch.setattr("slic.utils.time_runner.sleep", lambda x: None)
|
|
monkeypatch.setattr("slic.utils.time_runner.tqdm_mod", lambda *a, **k: DummyBar())
|
|
|
|
# Should not trigger yet
|
|
run_in(timedelta(seconds=20), fexample, "not yet")
|
|
assert "not yet" not in triggered
|
|
|
|
# After advancing the clock : should now trigger
|
|
fake_dt.advance(25)
|
|
run_in(timedelta(seconds=20), fexample, "now OK 1")
|
|
assert "now OK 1" in triggered
|
|
|
|
# Negative timedelta simulates a past schedule : should not run
|
|
fake_dt.advance(-100)
|
|
run_in(timedelta(seconds=-1), fexample, "past")
|
|
assert "past" not in triggered
|
|
|
|
out = capsys.readouterr().out
|
|
print(out)
|
|
|
|
|
|
def test_run_later(monkeypatch, capsys):
|
|
triggered = []
|
|
|
|
def fexample(label):
|
|
print(f"run_later triggered: {label}")
|
|
triggered.append(label)
|
|
|
|
base = datetime(2025, 1, 1, 14, 0, 0)
|
|
fake_dt = FakeDatetime(base)
|
|
|
|
# Patch for control over time and sleeping
|
|
monkeypatch.setattr("slic.utils.time_runner.datetime", fake_dt)
|
|
monkeypatch.setattr("slic.utils.time_runner.sleep", lambda x: None)
|
|
monkeypatch.setattr("slic.utils.time_runner.tqdm_mod", lambda *a, **k: DummyBar())
|
|
|
|
# Run using float seconds : should not trigger yet
|
|
run_later(15, fexample, "15s later")
|
|
assert "15s later" not in triggered
|
|
|
|
# After time passes : should trigger
|
|
fake_dt.advance(20)
|
|
run_later(15, fexample, "now OK 2")
|
|
assert "now OK 2" in triggered
|
|
|
|
# Timedelta in the past : should not run
|
|
run_later(timedelta(seconds=-5), fexample, "past")
|
|
assert "past" not in triggered
|
|
|
|
# Test run_later with datetime input
|
|
target = base + timedelta(seconds=10)
|
|
run_later(target, fexample, "datetime later") # not yet
|
|
fake_dt.advance(11)
|
|
run_later(target, fexample, "datetime now OK") # should now run
|
|
assert "datetime now OK" in triggered
|
|
|
|
out = capsys.readouterr().out
|
|
print(out)
|