chuanqi/tests/test_scheduler.py

# tests/test_scheduler.py
import lupa.luajit21 as lj
from apilot.stubs.scheduler import Scheduler, install_scheduler

def test_interval_and_oneshot():
    s = Scheduler()
    hits = []
    s.schedule(lambda: hits.append("i"), 0.05)
    s.delay(lambda: hits.append("d"), 0.1)
    for _ in range(30):
        s.tick(0.05)
    assert hits.count("i") >= 2
    assert hits.count("d") == 1

def test_lua_side_globals():
    rt = lj.LuaRuntime()
    from apilot.stubs.noop import install_noop_globals
    install_noop_globals(rt)
    s = Scheduler()
    install_scheduler(rt, s)
    rt.execute("global = global or {}; global.Scheduler = SL_getScheduler()")
    # install_scheduler 需暴露一个 Lua 可达引用; 简化: 直接用注入的全局函数
    rt.execute("n = 0; id = PerformWithDelayGlobal(function() n = n + 1 end, 0)")
    for _ in range(3):
        s.tick(0.05)
    assert rt.eval("n") == 1

def test_schedule_repeats():
    """Schedule 第三参为 nil 时按循环语义注册。"""
    rt = lj.LuaRuntime()
    install_scheduler(rt, Scheduler())
    rt.execute("n = 0; id = Schedule(function() n = n + 1 end, 0.05, nil)")
    for _ in range(3):
        s = rt.globals()["SL_getScheduler"]()
        s.tick(0.05)
    assert rt.eval("n") == 3

def test_schedule_two_arg_repeats():
    """Schedule 两参调用(第三参缺省)按循环语义注册。"""
    rt = lj.LuaRuntime()
    install_scheduler(rt, Scheduler())
    rt.execute("n = 0; id = Schedule(function() n = n + 1 end, 0.05)")
    for _ in range(3):
        s = rt.globals()["SL_getScheduler"]()
        s.tick(0.05)
    assert rt.eval("n") == 3

def test_cross_unschedule_within_tick():
    """A 到期回调里 UnSchedule(B) 不崩, 且 B 不再触发。"""
    rt = lj.LuaRuntime()
    s = Scheduler()
    install_scheduler(rt, s)
    rt.execute(
        "hits = 0; "
        "id_b = Schedule(function() hits = hits + 1 end, 0.05); "
        "Schedule(function() UnSchedule(id_b) end, 0.05)"
    )
    s.tick(0.05)  # A、B 同时到期: A 先注册先遍历, 回调里取消 B
    s.tick(0.05)
    s.tick(0.05)
    assert rt.eval("hits") == 1

def test_unschedule():
    s = Scheduler()
    hits = []
    eid = s.schedule(lambda: hits.append(1), 0.05)
    s.unschedule(eid)
    s.tick(0.05)
    assert hits == []