This commit is contained in:
+117
-76
@@ -4,84 +4,44 @@ from datetime import datetime, timedelta
|
||||
|
||||
from slic.utils.run_later import run_at, run_in, run_later, today, tomorrow, yesterday
|
||||
|
||||
'''
|
||||
# Dummy progress bar to disable actual waiting
|
||||
class DummyBar:
|
||||
def __enter__(self): return self
|
||||
def __exit__(self, *a): pass
|
||||
def set(self, val): pass
|
||||
|
||||
@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())
|
||||
|
||||
def test_run_at(patch_time):
|
||||
triggered = []
|
||||
def fexample(label): triggered.append(label)
|
||||
|
||||
with time_machine.travel("2025-01-01 06:00:00", tick=False) as traveller:
|
||||
# Too early
|
||||
run_at(today(6, 30), fexample, "too early")
|
||||
assert "too early" not in triggered
|
||||
|
||||
# On time
|
||||
traveller.move_to("2025-01-01 06:30:00")
|
||||
run_at(today(6, 30), fexample, "on time")
|
||||
assert "on time" in triggered
|
||||
|
||||
# Later
|
||||
traveller.move_to("2025-01-01 07:00:00")
|
||||
run_at(today(6, 30), fexample, "later")
|
||||
assert "later" in triggered
|
||||
|
||||
def test_run_in(patch_time):
|
||||
triggered = []
|
||||
def fexample(label): triggered.append(label)
|
||||
|
||||
with time_machine.travel("2025-01-01 12:00:00", tick=False) as traveller:
|
||||
delay = timedelta(seconds=60)
|
||||
|
||||
# Too early
|
||||
run_in(delay, fexample, "too early")
|
||||
assert "too early" not in triggered
|
||||
|
||||
# On time
|
||||
traveller.move_to("2025-01-01 12:01:00")
|
||||
run_in(delay, fexample, "on time")
|
||||
assert "on time" in triggered
|
||||
|
||||
# Later
|
||||
traveller.move_to("2025-01-01 12:02:00")
|
||||
run_in(delay, fexample, "later")
|
||||
assert "later" in triggered
|
||||
|
||||
def test_run_later(patch_time):
|
||||
triggered = []
|
||||
def fexample(label): triggered.append(label)
|
||||
|
||||
with time_machine.travel("2025-01-01 14:00:00", tick=False) as traveller:
|
||||
# Too early
|
||||
run_later(60, fexample, "too early")
|
||||
assert "too early" not in triggered
|
||||
|
||||
# On time
|
||||
traveller.move_to("2025-01-01 14:01:00")
|
||||
run_later(60, fexample, "on time")
|
||||
assert "on time" in triggered
|
||||
|
||||
# Later
|
||||
traveller.move_to("2025-01-01 14:02:00")
|
||||
run_later(60, fexample, "later")
|
||||
assert "later" in triggered
|
||||
'''
|
||||
|
||||
import pytest
|
||||
import time_machine
|
||||
from datetime import datetime, timedelta
|
||||
from slic.utils.run_later import run_at
|
||||
from slic.utils.run_later import MatMulTrick
|
||||
from datetime import datetime
|
||||
|
||||
|
||||
# Test MatMulTrick
|
||||
|
||||
def original_func(*args, **kwargs):
|
||||
return ("called", args, kwargs)
|
||||
|
||||
def test_init_and_repr():
|
||||
wrapped = MatMulTrick(original_func)
|
||||
assert wrapped.func is original_func
|
||||
assert repr(wrapped) == repr(original_func)
|
||||
|
||||
def test_call():
|
||||
wrapped = MatMulTrick(original_func)
|
||||
result = wrapped(1, 2, a=3)
|
||||
assert result == ("called", (1, 2), {'a': 3})
|
||||
|
||||
def test_matmul_with_string():
|
||||
wrapped = MatMulTrick(original_func)
|
||||
result = wrapped @ "6:30"
|
||||
assert result == ("called", (6, 30), {})
|
||||
|
||||
def test_matmul_with_list():
|
||||
wrapped = MatMulTrick(original_func)
|
||||
result = wrapped @ [6, 30]
|
||||
assert result == ("called", (6, 30), {})
|
||||
|
||||
def test_matmul_with_single_int():
|
||||
wrapped = MatMulTrick(original_func)
|
||||
result = wrapped @ 7
|
||||
assert result == ("called", (7,), {})
|
||||
|
||||
|
||||
# Test run functions
|
||||
|
||||
def fexample(label):
|
||||
print(f"Triggered: {label}")
|
||||
|
||||
@@ -107,4 +67,85 @@ def test_run_at_already_past(capsys):
|
||||
|
||||
assert "it is already past" in captured
|
||||
assert "ago" in captured
|
||||
assert "Triggered: B" not in captured
|
||||
assert "Triggered: B" not in captured
|
||||
|
||||
def test_run_in_future_triggered_and_logs(capsys):
|
||||
with time_machine.travel(datetime(2025, 1, 1, 12, 0, 0), tick=True):
|
||||
run_in(timedelta(seconds=0.2), fexample, "IN")
|
||||
|
||||
captured = capsys.readouterr().out
|
||||
|
||||
assert "will run at" in captured
|
||||
assert "this will be in" in captured
|
||||
assert "Triggered: IN" in captured
|
||||
|
||||
|
||||
def test_run_in_past(capsys):
|
||||
with time_machine.travel(datetime(2025, 1, 1, 12, 0, 0), tick=True):
|
||||
run_in(-timedelta(seconds=2), fexample, "IN PAST")
|
||||
|
||||
captured = capsys.readouterr().out
|
||||
|
||||
assert "it is already past" in captured
|
||||
assert "Triggered: IN PAST" not in captured
|
||||
|
||||
from slic.utils.run_later import run_later
|
||||
|
||||
|
||||
def test_run_later_with_seconds(capsys):
|
||||
with time_machine.travel(datetime(2025, 1, 1, 14, 0, 0), tick=True):
|
||||
run_later(0.2, fexample, "LATER")
|
||||
|
||||
captured = capsys.readouterr().out
|
||||
|
||||
assert "will run at" in captured
|
||||
assert "Triggered: LATER" in captured
|
||||
|
||||
|
||||
def test_run_later_with_past_datetime(capsys):
|
||||
with time_machine.travel(datetime(2025, 1, 1, 14, 0, 0), tick=True):
|
||||
past_time = datetime(2025, 1, 1, 13, 59, 59)
|
||||
run_later(past_time, fexample, "TOO LATE")
|
||||
|
||||
captured = capsys.readouterr().out
|
||||
|
||||
assert "it is already past" in captured
|
||||
assert "Triggered: TOO LATE" not in captured
|
||||
|
||||
|
||||
# Test today/yesterday and tomorrow functions
|
||||
|
||||
def test_today_basic():
|
||||
with time_machine.travel("2025-01-01 10:00:00", tick=False):
|
||||
val = today(12, 30)
|
||||
assert val == datetime(2025, 1, 1, 12, 30)
|
||||
|
||||
|
||||
def test_tomorrow_basic():
|
||||
with time_machine.travel("2025-01-01 10:00:00", tick=False):
|
||||
val = tomorrow(12, 30)
|
||||
assert val == datetime(2025, 1, 2, 12, 30)
|
||||
|
||||
|
||||
def test_yesterday_basic():
|
||||
with time_machine.travel("2025-01-01 10:00:00", tick=False):
|
||||
val = yesterday(12, 30)
|
||||
assert val == datetime(2024, 12, 31, 12, 30)
|
||||
|
||||
|
||||
def test_today_matmul_string():
|
||||
with time_machine.travel("2025-01-01 00:00:00", tick=False):
|
||||
val = today @ "15:45"
|
||||
assert val == datetime(2025, 1, 1, 15, 45)
|
||||
|
||||
|
||||
def test_tomorrow_matmul_list():
|
||||
with time_machine.travel("2025-01-01 00:00:00", tick=False):
|
||||
val = tomorrow @ [6, 0]
|
||||
assert val == datetime(2025, 1, 2, 6, 0)
|
||||
|
||||
|
||||
def test_yesterday_matmul_single():
|
||||
with time_machine.travel("2025-01-01 00:00:00", tick=False):
|
||||
val = yesterday @ 7
|
||||
assert val == datetime(2024, 12, 31, 7, 0)
|
||||
|
||||
Reference in New Issue
Block a user