yumao
6ea2a4e4c6
- 新增常驻调度器 daily/scheduler.py + run-scheduler.ps1（定时生成/推送） - 新增 daily/bridge_manager.py：Windows 兼容的 Cursor SDK 桥接 - 新增 skills/daily-featured-pick 首推 Skill 与叙事轴/去重逻辑 - 新闻抓取窗口、GitHub 搜索、企微 delta 模式等多项改进 - 补充设计文档与 superpowers 计划/规范 - 新增对应测试（scheduler、featured_pick、github_search、news_fetch_window 等） Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
119 lines
3.7 KiB
Python
119 lines
3.7 KiB
Python
"""Tests for daily.scheduler."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import unittest
|
|
from datetime import datetime
|
|
from zoneinfo import ZoneInfo
|
|
|
|
from daily.scheduler import (
|
|
ClockTime,
|
|
SchedulerState,
|
|
next_occurrence_after,
|
|
parse_hhmm,
|
|
plan_next_action,
|
|
)
|
|
|
|
|
|
class ParseHhmmTests(unittest.TestCase):
|
|
def test_parse(self):
|
|
t = parse_hhmm("08:50")
|
|
self.assertEqual((t.hour, t.minute), (8, 50))
|
|
|
|
def test_invalid(self):
|
|
with self.assertRaises(ValueError):
|
|
parse_hhmm("25:00")
|
|
|
|
|
|
class PlanNextActionTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.tz = ZoneInfo("Asia/Shanghai")
|
|
self.gen = ClockTime(8, 50)
|
|
self.push = ClockTime(9, 0)
|
|
|
|
def test_before_generate_waits_for_generate(self):
|
|
now = datetime(2026, 7, 9, 8, 30, tzinfo=self.tz)
|
|
run_at, action = plan_next_action(
|
|
now=now,
|
|
tz=self.tz,
|
|
state=SchedulerState(),
|
|
generate_at=self.gen,
|
|
push_at=self.push,
|
|
)
|
|
self.assertEqual(action, "generate")
|
|
self.assertEqual(run_at.hour, 8)
|
|
self.assertEqual(run_at.minute, 50)
|
|
|
|
def test_after_generate_before_push_waits_for_push(self):
|
|
now = datetime(2026, 7, 9, 8, 55, tzinfo=self.tz)
|
|
state = SchedulerState(last_generate_date="2026-07-09")
|
|
run_at, action = plan_next_action(
|
|
now=now,
|
|
tz=self.tz,
|
|
state=state,
|
|
generate_at=self.gen,
|
|
push_at=self.push,
|
|
)
|
|
self.assertEqual(action, "push")
|
|
self.assertEqual(run_at.hour, 9)
|
|
|
|
def test_catch_up_generate_when_started_late(self):
|
|
now = datetime(2026, 7, 9, 8, 55, tzinfo=self.tz)
|
|
run_at, action = plan_next_action(
|
|
now=now,
|
|
tz=self.tz,
|
|
state=SchedulerState(),
|
|
generate_at=self.gen,
|
|
push_at=self.push,
|
|
)
|
|
self.assertEqual(action, "generate")
|
|
self.assertEqual(run_at, now)
|
|
|
|
def test_next_day_after_both_done(self):
|
|
now = datetime(2026, 7, 9, 10, 0, tzinfo=self.tz)
|
|
state = SchedulerState(last_generate_date="2026-07-09", last_push_date="2026-07-09")
|
|
run_at, action = plan_next_action(
|
|
now=now,
|
|
tz=self.tz,
|
|
state=state,
|
|
generate_at=self.gen,
|
|
push_at=self.push,
|
|
)
|
|
self.assertEqual(action, "generate")
|
|
self.assertEqual(run_at.date().isoformat(), "2026-07-10")
|
|
|
|
def test_evening_start_waits_for_tomorrow_generate(self):
|
|
now = datetime(2026, 7, 9, 20, 35, tzinfo=self.tz)
|
|
run_at, action = plan_next_action(
|
|
now=now,
|
|
tz=self.tz,
|
|
state=SchedulerState(),
|
|
generate_at=self.gen,
|
|
push_at=self.push,
|
|
)
|
|
self.assertEqual(action, "generate")
|
|
self.assertEqual(run_at.date().isoformat(), "2026-07-10")
|
|
self.assertEqual((run_at.hour, run_at.minute), (8, 50))
|
|
|
|
def test_catch_up_push_when_generate_done(self):
|
|
now = datetime(2026, 7, 9, 20, 35, tzinfo=self.tz)
|
|
state = SchedulerState(last_generate_date="2026-07-09")
|
|
run_at, action = plan_next_action(
|
|
now=now,
|
|
tz=self.tz,
|
|
state=state,
|
|
generate_at=self.gen,
|
|
push_at=self.push,
|
|
)
|
|
self.assertEqual(action, "push")
|
|
self.assertEqual(run_at, now)
|
|
|
|
|
|
class NextOccurrenceTests(unittest.TestCase):
|
|
def test_tomorrow_when_past(self):
|
|
tz = ZoneInfo("Asia/Shanghai")
|
|
now = datetime(2026, 7, 9, 10, 0, tzinfo=tz)
|
|
nxt = next_occurrence_after(ClockTime(8, 50), tz, now)
|
|
self.assertEqual(nxt.date().isoformat(), "2026-07-10")
|
|
self.assertEqual((nxt.hour, nxt.minute), (8, 50))
|