"""core/scheduler.py 服务层测试(DESIGN §8.5)—— 补审查 Top3 缺口(此前零测试)。 覆盖两层: - 纯逻辑(无 DB):cron 校验、next_run 时区折算(墙钟语义)、run message 包装 - DB 级(仅当显式设 ZCBOT_TEST_DB_URL 才跑;只插/删测试专属 user 的行): create/claim 推进 next_run(不重复认领)、disabled/软删/过期不认领、 record_result 连续失败自动停用 端到端(守护循环认领 → 真跑 agent → LLM 回复)仍走 scripts/smoke_scheduler.py (需起服务 + 真实调用,不进单测)。 """ from __future__ import annotations import os import unittest import uuid from datetime import datetime, timedelta, timezone def _test_db_ready() -> bool: """DB 级测试只认显式 `ZCBOT_TEST_DB_URL`,绝不回退 .env 的 ZCBOT_DB_URL。 教训(2026-07-23 实锤):.env 的 ZCBOT_DB_URL 可能经隧道指向**生产库**(本次 即 127.0.0.1:6012 → 生产 PG),测试若沿用它,插入的 enabled 到点 job 会被生产 实例的调度守护认领并真跑 agent。测试库须显式指定: set ZCBOT_TEST_DB_URL=postgresql+psycopg://user:pass@127.0.0.1:5432/zcbot_test (建库后 `alembic upgrade head` 一次);未设 → 整组 skip。 """ url = os.environ.get("ZCBOT_TEST_DB_URL", "").strip() if not url: return False os.environ["ZCBOT_DB_URL"] = url # 本测试进程内覆盖,engine 单例随之指向测试库 return True from core import scheduler # noqa: E402 (纯逻辑部分不碰 DB,导入安全) try: if not _test_db_ready(): raise RuntimeError("ZCBOT_TEST_DB_URL 未设") from core.storage import session_scope from core.storage.models import ScheduledJob, User with session_scope() as _s: _s.execute(__import__("sqlalchemy").select(1)) _DB_OK = True except Exception: _DB_OK = False class SchedulerPureTests(unittest.TestCase): def test_validate_cron(self): scheduler.validate_cron("0 8 * * *") scheduler.validate_cron("*/5 * * * *") for bad in ("", "not a cron", "99 99 * * *", "* * * *"): with self.assertRaises(ValueError): scheduler.validate_cron(bad) def test_compute_next_run_wall_clock_tz(self): """'0 8 * * *' 是 job 时区的早 8 点,不是 UTC 8 点(§8.5 时区坑)。""" # 2026-01-05 00:00 UTC = 上海 08:00 —— 下一个上海 08:00 是 1 月 6 日 after = datetime(2026, 1, 5, 0, 0, tzinfo=timezone.utc) nxt = scheduler.compute_next_run("0 8 * * *", "Asia/Shanghai", after=after) self.assertEqual(nxt.tzinfo, timezone.utc) self.assertEqual((nxt.year, nxt.month, nxt.day, nxt.hour), (2026, 1, 6, 0)) # 同一时刻按 UTC 时区算,则当天 08:00 UTC nxt_utc = scheduler.compute_next_run("0 8 * * *", "UTC", after=after) self.assertEqual((nxt_utc.day, nxt_utc.hour), (5, 8)) def test_build_run_message_wraps_prompt(self): msg = scheduler.build_run_message( {"name": "早安简报", "prompt": "回一句早安", "tz": "Asia/Shanghai"} ) self.assertIn("[定时任务「早安简报」自动触发", msg) self.assertIn("回一句早安", msg) @unittest.skipUnless(_DB_OK, "ZCBOT_TEST_DB_URL 未设或测试库不可达,跳过 DB 级测试(绝不用 .env 的库)") class SchedulerDbTests(unittest.TestCase): @classmethod def setUpClass(cls): cls.uid = uuid.uuid4() with session_scope() as s: s.add(User(user_id=cls.uid, email=f"test-scheduler-{cls.uid.hex[:8]}@invalid.local")) @classmethod def tearDownClass(cls): from sqlalchemy import delete with session_scope() as s: s.execute(delete(ScheduledJob).where(ScheduledJob.user_id == cls.uid)) s.execute(delete(User).where(User.user_id == cls.uid)) def _mk_job(self, **kw) -> uuid.UUID: d = scheduler.create_job( self.uid, name=kw.pop("name", "t-job"), prompt=kw.pop("prompt", "p"), cron=kw.pop("cron", "0 8 * * *"), **kw, ) return uuid.UUID(d["job_id"]) def _set(self, jid: uuid.UUID, **values) -> None: from sqlalchemy import update with session_scope() as s: s.execute(update(ScheduledJob).where(ScheduledJob.job_id == jid).values(**values)) def _row(self, jid: uuid.UUID) -> ScheduledJob: with session_scope() as s: return s.get(ScheduledJob, jid) # 认领一律用过去纪元:claim_due_jobs 是全库扫描,若用真实 now 会把真实用户 # 到点的 job 一并认领(推进 next_run_at 但不执行 = 吃掉一次真实触发)。 # 测试 job 的 next_run_at 都摆在 2000 年,真实 job(next_run 恒近未来)永不命中。 EPOCH_DUE = datetime(2000, 1, 1, tzinfo=timezone.utc) EPOCH_NOW = datetime(2000, 1, 2, tzinfo=timezone.utc) def _claim_ids(self) -> set: return {snap["job_id"] for snap in scheduler.claim_due_jobs(now=self.EPOCH_NOW)} def test_create_sets_next_run_and_validates(self): jid = self._mk_job() row = self._row(jid) self.assertIsNotNone(row.next_run_at) self.assertTrue(row.enabled) with self.assertRaises(scheduler.JobError): scheduler.create_job(self.uid, name="", prompt="p", cron="0 8 * * *") with self.assertRaises(ValueError): scheduler.create_job(self.uid, name="x", prompt="p", cron="bad cron") def test_claim_advances_next_run_no_double_claim(self): jid = self._mk_job(name="due-job") self._set(jid, next_run_at=self.EPOCH_DUE) self.assertIn(jid, self._claim_ids()) # 到点 → 认领 row = self._row(jid) self.assertGreater(row.next_run_at, self.EPOCH_NOW) # 已按 cron 推进到 now 之后 self.assertNotIn(jid, self._claim_ids()) # 同一到点不重复认领 def test_disabled_and_deleted_not_claimed(self): jid = self._mk_job(name="off-job") self._set(jid, next_run_at=self.EPOCH_DUE) scheduler.set_enabled(self.uid, str(jid), False) self.assertNotIn(jid, self._claim_ids()) jid2 = self._mk_job(name="del-job") self._set(jid2, next_run_at=self.EPOCH_DUE) scheduler.cancel_job(self.uid, str(jid2)) self.assertNotIn(jid2, self._claim_ids()) self.assertIsNotNone(self._row(jid2).deleted_at) # 软删,不物理删 def test_expired_job_auto_disabled(self): jid = self._mk_job(name="expired-job") self._set(jid, next_run_at=self.EPOCH_DUE, expires_at=self.EPOCH_DUE + timedelta(hours=1)) self.assertNotIn(jid, self._claim_ids()) # 过期不返回 row = self._row(jid) self.assertFalse(row.enabled) self.assertEqual(row.last_status, "expired") def test_record_result_failure_streak_auto_disable(self): jid = self._mk_job(name="fail-job") for i in range(scheduler.FAILURE_DISABLE_THRESHOLD - 1): scheduler.record_result(jid, status="error", task_id=None, error=f"boom {i}") row = self._row(jid) self.assertTrue(row.enabled) # 还差一次,不停 # 一次 ok 清零计数 scheduler.record_result(jid, status="ok", task_id=None) self.assertEqual(self._row(jid).consecutive_failures, 0) # 连续失败到阈值 → 自动停用,last_error 带说明 for i in range(scheduler.FAILURE_DISABLE_THRESHOLD): scheduler.record_result(jid, status="error", task_id=None, error="boom") row = self._row(jid) self.assertFalse(row.enabled) self.assertIn("已自动停用", row.last_error or "") def test_skipped_does_not_touch_counters(self): jid = self._mk_job(name="skip-job") scheduler.record_result(jid, status="skipped", task_id=None, error="busy") row = self._row(jid) self.assertEqual(row.run_count or 0, 0) self.assertEqual(row.consecutive_failures or 0, 0) self.assertEqual(row.last_status, "skipped") if __name__ == "__main__": unittest.main()