tests/test_follow_scheduler.py
Rama master · réplica de lectura del repositorio original.
import tempfile
import unittest
from datetime import datetime, timedelta, timezone
from pathlib import Path
from peach.follow_scheduler import (
MAX_BACKOFF_FACTOR,
MAX_INTERVAL_MINUTES,
MIN_INTERVAL_MINUTES,
FollowScheduleConfig,
FollowScheduleStore,
FollowUpdateScheduler,
)
class FollowScheduleStoreTests(unittest.TestCase):
def test_defaults_are_enabled_hourly_and_updates_are_atomic(self):
with tempfile.TemporaryDirectory() as temporary:
store = FollowScheduleStore(Path(temporary))
self.assertEqual(store.load(), FollowScheduleConfig(True, 60))
store.save(FollowScheduleConfig(False, 180))
self.assertEqual(store.load(), FollowScheduleConfig(False, 180))
def test_unbounded_intervals_are_rejected(self):
with self.assertRaises(ValueError):
FollowScheduleStore.validate(True, 14)
with self.assertRaises(ValueError):
FollowScheduleStore.validate(True, 10081)
with self.assertRaises(ValueError):
FollowScheduleStore.validate("false", 60)
class FollowUpdateSchedulerTests(unittest.TestCase):
def test_restart_preserves_failed_checks_and_the_backoff_deadline(self):
failure = lambda: {'checked': 1, 'results': [{'ok': False}]}
with tempfile.TemporaryDirectory() as temporary:
first = FollowUpdateScheduler(Path(temporary), failure, available=True)
first.start()
try:
first._run()
first._run()
previous = first.status()
finally:
first.stop()
second = FollowUpdateScheduler(Path(temporary), failure, available=True)
second.start()
try:
current = second.status()
self.assertEqual(current['consecutive_failures'], 2)
self.assertEqual(current['last_error'], previous['last_error'])
self.assertEqual(current['failing_since'], previous['failing_since'])
self.assertEqual(current['next_run_at'], previous['next_run_at'])
second.run_check = lambda: {'checked': 1, 'results': [{'ok': True}]}
second._run()
finally:
second.stop()
third = FollowUpdateScheduler(Path(temporary), failure, available=True)
self.assertEqual(third.status()['consecutive_failures'], 0)
self.assertIsNone(third.status()['last_error'])
self.assertIsNone(third.status()['failing_since'])
def test_installation_reservation_skips_the_tick_without_recording_a_failure(self):
from peach.update_tasks import UpdateBusy
def blocked():
raise UpdateBusy('正在安装')
with tempfile.TemporaryDirectory() as temporary:
scheduler = FollowUpdateScheduler(Path(temporary), blocked, available=True)
with self.assertNoLogs('peach', level='ERROR'):
scheduler._run()
status = scheduler.status()
self.assertEqual(status['consecutive_failures'], 0)
self.assertIsNone(status['last_error'])
self.assertIsNotNone(status['last_skipped_at'])
def test_interval_job_waits_one_interval_and_can_be_disabled(self):
with tempfile.TemporaryDirectory() as temporary:
scheduler = FollowUpdateScheduler(Path(temporary), lambda: {"ok": True}, available=True)
self.addCleanup(scheduler.stop)
scheduler.start()
status = scheduler.status()
self.assertTrue(status["enabled"])
self.assertIsNotNone(status["next_run_at"])
status = scheduler.update(enabled=False, interval_minutes=60)
self.assertFalse(status["enabled"])
self.assertIsNone(status["next_run_at"])
def test_run_status_reports_checked_sources_and_additions(self):
result = {"ok": True, "checked": 2, "results": [
{"ok": True, "added": 3}, {"ok": False, "added": 0},
]}
with tempfile.TemporaryDirectory() as temporary:
scheduler = FollowUpdateScheduler(Path(temporary), lambda: result, available=True)
scheduler._run()
status = scheduler.status()
self.assertEqual(status["last_checked"], 2)
self.assertEqual(status["last_added"], 3)
self.assertEqual(status["last_error"], "1 个来源检查失败")
self.assertIsNotNone(status["last_finished_at"])
def test_a_busy_lock_is_a_skip_not_a_failure(self):
"""互斥不是故障：手动检查正在跑的时候，自动这一轮什么也没做。
这里抛异常的话，界面会把它显示成「上次失败」，还连带触发退避——真正的
失败反而被这种噪声盖住。"""
with tempfile.TemporaryDirectory() as temporary:
scheduler = FollowUpdateScheduler(
Path(temporary), lambda: {"ok": False, "busy": True,
"checked": 0, "results": []},
available=True)
scheduler._run()
status = scheduler.status()
self.assertIsNone(status["last_error"])
self.assertIsNotNone(status["last_skipped_at"])
self.assertEqual(status["consecutive_failures"], 0)
# 这一轮既没检查也没完成，所以不留「上次运行」的时间戳：
# 留了界面就会把一次没发生的运行显示成「上次完成」。
self.assertIsNone(status["last_started_at"])
self.assertIsNone(status["last_finished_at"])
self.assertEqual(status["last_checked"], 0)
def test_consecutive_failures_back_off_and_a_good_run_resets_them(self):
outcomes = [
{"ok": True, "checked": 1, "results": [{"ok": False, "added": 0}]},
RuntimeError("上游断了"),
{"ok": True, "checked": 1, "results": [{"ok": True, "added": 2}]},
]
def run_check():
outcome = outcomes.pop(0)
if isinstance(outcome, Exception):
raise outcome
return outcome
with tempfile.TemporaryDirectory() as temporary:
scheduler = FollowUpdateScheduler(Path(temporary), run_check,
available=True)
interval = scheduler.config.interval_minutes
# 第一次失败照原间隔重试：多数失败只是一次抖动。
self.assertEqual(scheduler.backoff_minutes(1), interval)
self.assertEqual(scheduler.backoff_minutes(2), interval * 2)
self.assertEqual(scheduler.backoff_minutes(3), interval * 4)
self.assertEqual(scheduler.backoff_minutes(99),
min(interval * MAX_BACKOFF_FACTOR,
MAX_INTERVAL_MINUTES))
scheduler._run()
first = scheduler.status()
self.assertEqual(first["consecutive_failures"], 1)
self.assertEqual(first["failing_since"], first["last_started_at"])
with self.assertLogs("peach.follow_scheduler", level="ERROR") as captured:
scheduler._run()
self.assertIn("上游断了", "\n".join(captured.output))
status = scheduler.status()
self.assertEqual(status["consecutive_failures"], 2)
self.assertEqual(status["last_error"], "上游断了")
# 连续失败的起点停在第一轮，不跟着每一轮往后挪。
self.assertEqual(status["failing_since"], first["failing_since"])
scheduler._run()
status = scheduler.status()
self.assertEqual(status["consecutive_failures"], 0)
self.assertIsNone(status["last_error"])
self.assertIsNone(status["failing_since"])
self.assertEqual(status["last_added"], 2)
def test_run_now_runs_the_scheduled_job_and_settles_its_failure(self):
import threading
finished = threading.Event()
outcomes = [{"ok": True, "checked": 1, "results": [{"ok": False}]},
{"ok": True, "checked": 1, "results": [{"ok": True, "added": 1}]}]
def run_check():
return outcomes.pop(0)
with tempfile.TemporaryDirectory() as temporary:
scheduler = FollowUpdateScheduler(Path(temporary), run_check, available=True)
self.addCleanup(scheduler.stop)
with self.assertRaises(ValueError):
scheduler.run_now()
scheduler._run()
self.assertEqual(scheduler.status()["last_error"], "1 个来源检查失败")
original = scheduler._run
def tracked():
original()
finished.set()
scheduler._run = tracked
scheduler.start()
# 提前标成运行中，读状态的人不会在调度线程领走之前再看到旧的失败。
self.assertTrue(scheduler.run_now()["running"])
self.assertTrue(finished.wait(10))
status = scheduler.status()
self.assertEqual((status["running"], status["last_error"], status["consecutive_failures"]),
(False, None, 0))
self.assertIsNotNone(status["next_run_at"])
def test_run_now_is_refused_without_a_running_job(self):
with tempfile.TemporaryDirectory() as temporary:
reader = FollowUpdateScheduler(Path(temporary), lambda: {"ok": True}, available=False)
with self.assertRaises(ValueError):
reader.run_now()
writer = FollowUpdateScheduler(Path(temporary), lambda: {"ok": True}, available=True)
self.addCleanup(writer.stop)
writer.start()
writer.update(enabled=False, interval_minutes=60)
with self.assertRaises(ValueError):
writer.run_now()
self.assertFalse(writer.status()["running"])
def test_backoff_pushes_the_next_run_past_the_plain_interval(self):
with tempfile.TemporaryDirectory() as temporary:
scheduler = FollowUpdateScheduler(
Path(temporary),
lambda: {"ok": True, "checked": 1,
"results": [{"ok": False, "added": 0}]},
available=True)
self.addCleanup(scheduler.stop)
scheduler.update(enabled=True, interval_minutes=MIN_INTERVAL_MINUTES)
scheduler.start()
planned = datetime.now(timezone.utc) + timedelta(
minutes=MIN_INTERVAL_MINUTES)
scheduler._run()
scheduler._run()
next_run = scheduler.status()["next_run_at"]
self.assertIsNotNone(next_run)
self.assertGreater(datetime.fromisoformat(next_run), planned,
"连续失败后应当推后，而不是照原间隔继续敲上游")
def test_a_tick_shorter_than_the_period_runs_often_and_backs_off_from_the_tick(self):
"""订阅源：设置里的间隔是每个源的周期，调度按更短的一跳看谁到期（ADR-0134）。"""
with tempfile.TemporaryDirectory() as temporary:
scheduler = FollowUpdateScheduler(
Path(temporary), lambda: {"ok": True, "checked": 0, "results": []},
available=True, default=FollowScheduleConfig(True, 360), tick_minutes=30)
self.addCleanup(scheduler.stop)
before = datetime.now(timezone.utc)
scheduler.start()
status = scheduler.status()
self.assertEqual(status["interval_minutes"], 360)
self.assertLess(datetime.fromisoformat(status["next_run_at"]),
before + timedelta(minutes=31))
self.assertEqual(scheduler.backoff_minutes(2), 60)
def test_an_idle_tick_keeps_the_last_failure_on_record(self):
"""没有到期源的一跳什么也没做：上一轮的失败、连续失败计数和上次运行时间都留着。"""
outcomes = [{"ok": True, "checked": 1, "results": [{"ok": False, "added": 0}]},
{"ok": True, "idle": True, "checked": 0, "results": []}]
with tempfile.TemporaryDirectory() as temporary:
scheduler = FollowUpdateScheduler(Path(temporary), lambda: outcomes.pop(0),
available=True, tick_minutes=30)
scheduler._run()
failed = scheduler.status()
scheduler._run()
status = scheduler.status()
self.assertEqual(status["last_error"], "1 个来源检查失败")
self.assertEqual(status["consecutive_failures"], 1)
self.assertEqual((status["last_started_at"], status["last_finished_at"], status["last_checked"]),
(failed["last_started_at"], failed["last_finished_at"], 1))
self.assertFalse(status["running"])
def test_reader_never_starts_or_accepts_schedule_changes(self):
with tempfile.TemporaryDirectory() as temporary:
scheduler = FollowUpdateScheduler(Path(temporary), lambda: {}, available=False)
scheduler.start()
self.assertFalse(scheduler.status()["available"])
with self.assertRaises(ValueError):
scheduler.update(enabled=True, interval_minutes=60)