kairee
5c2ab7b187
Implement P1: seven-metric view snapshots, relation/stat fan snapshots, failure placeholders, resume by daily completion markers, Cookie login warnings, daily SQLite backups, Shanghai daily scheduling, and CLI snapshot/auth-check commands. Real read-only run captured 795/795 videos successfully over 1124s with a verified daily backup.
142 lines
3.6 KiB
Python
142 lines
3.6 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""内置调度器测试（离线）: uv run python tests/test_scheduler.py"""
|
|
|
|
import os
|
|
import sys
|
|
import threading
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
sys.stdout.reconfigure(encoding="utf-8")
|
|
|
|
from reaction_board.scheduler import ScheduledTask, Scheduler # noqa: E402
|
|
|
|
|
|
def test_run_when_due_and_reschedule():
|
|
clock = [1_000.0]
|
|
calls = []
|
|
|
|
task = ScheduledTask("sync", 60, lambda: calls.append(clock[0]))
|
|
scheduler = Scheduler([task], clock=lambda: clock[0])
|
|
assert scheduler.run_due() and len(calls) == 1
|
|
assert task.next_run == 1_060.0
|
|
|
|
clock[0] = 1_059.0
|
|
assert scheduler.run_due() == []
|
|
assert len(calls) == 1
|
|
|
|
clock[0] = 1_060.0
|
|
results = scheduler.run_due()
|
|
assert len(results) == 1
|
|
assert calls == [1_000.0, 1_060.0]
|
|
assert task.next_run == 1_120.0
|
|
|
|
|
|
def test_task_error_does_not_block_other_tasks():
|
|
clock = [100.0]
|
|
calls = []
|
|
|
|
def boom():
|
|
raise RuntimeError("api down")
|
|
|
|
scheduler = Scheduler(
|
|
[
|
|
ScheduledTask("bad", 10, boom),
|
|
ScheduledTask("good", 10, lambda: calls.append("ok")),
|
|
],
|
|
clock=lambda: clock[0],
|
|
)
|
|
results = scheduler.run_due()
|
|
assert [r.ok for r in results] == [False, True]
|
|
assert results[0].error == "api down"
|
|
assert calls == ["ok"]
|
|
# 失败任务仍然被重新排期
|
|
assert all(t.next_run == 110.0 for t in scheduler.tasks)
|
|
|
|
|
|
def test_run_forever_with_stop_event():
|
|
clock = [0.0]
|
|
calls = []
|
|
|
|
def tick():
|
|
calls.append(clock[0])
|
|
clock[0] += 10
|
|
|
|
stop = threading.Event()
|
|
scheduler = Scheduler(
|
|
[ScheduledTask("tick", 10, tick)],
|
|
clock=lambda: clock[0],
|
|
sleeper=lambda seconds: None,
|
|
)
|
|
|
|
def stop_soon():
|
|
while len(calls) < 3:
|
|
pass
|
|
stop.set()
|
|
|
|
waiter = threading.Thread(target=stop_soon)
|
|
waiter.start()
|
|
scheduler.run_forever(stop_event=stop, poll_seconds=0.001)
|
|
waiter.join(timeout=2)
|
|
assert len(calls) >= 3
|
|
|
|
|
|
def test_max_iterations_and_validation():
|
|
calls = []
|
|
tick_clock = [0.0]
|
|
|
|
def next_time():
|
|
tick_clock[0] += 1.0
|
|
return tick_clock[0]
|
|
|
|
scheduler = Scheduler(
|
|
[ScheduledTask("t", 1, lambda: calls.append(1))],
|
|
clock=next_time,
|
|
sleeper=lambda seconds: None,
|
|
)
|
|
scheduler.run_forever(max_iterations=2)
|
|
assert len(calls) == 2
|
|
|
|
for tasks, kwargs in [
|
|
([], {}),
|
|
([ScheduledTask("bad", 0, lambda: None)], {}),
|
|
([ScheduledTask("t", 1, lambda: None)], {"poll_seconds": 0}),
|
|
]:
|
|
try:
|
|
s = Scheduler(tasks, sleeper=lambda seconds: None)
|
|
s.run_forever(max_iterations=1, **kwargs)
|
|
except ValueError:
|
|
pass
|
|
else:
|
|
raise AssertionError(f"应当失败: {tasks}, {kwargs}")
|
|
|
|
|
|
def test_next_run_fn_controls_reschedule():
|
|
clock = [1_000.0]
|
|
calls = []
|
|
task = ScheduledTask(
|
|
"daily", 86400, lambda: calls.append(clock[0]),
|
|
next_run_fn=lambda started: started + 123,
|
|
)
|
|
scheduler = Scheduler([task], clock=lambda: clock[0])
|
|
scheduler.run_due()
|
|
assert calls == [1_000.0]
|
|
assert task.next_run == 1_123.0
|
|
clock[0] = 1_122.0
|
|
assert scheduler.run_due() == []
|
|
clock[0] = 1_123.0
|
|
scheduler.run_due()
|
|
assert calls == [1_000.0, 1_123.0]
|
|
|
|
|
|
def main():
|
|
test_run_when_due_and_reschedule()
|
|
test_next_run_fn_controls_reschedule()
|
|
test_task_error_does_not_block_other_tasks()
|
|
test_run_forever_with_stop_event()
|
|
test_max_iterations_and_validation()
|
|
print("OK: 内置调度器测试全部通过")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|