JiuContinent
d7db84571d
feat: 添加飞书用户模块和订阅功能支持 - 新增feishu_users模块用于处理飞书用户身份验证和权限管理 - 新增subscriptions模块用于处理订阅相关功能 - 新增personalization模块用于个性化服务 - 在alembic迁移配置中注册新的模型模块 - 在API路由器中添加feishu_users和subscriptions路由 - 实现事件调度服务的改进，包括错误处理和状态更新优化 - 添加飞书命令处理的权限检查机制 - 实现飞书应用票据事件处理 - 改进审计日志记录功能 ```
106 lines
3.0 KiB
Python
106 lines
3.0 KiB
Python
from datetime import UTC, datetime, time
|
|
|
|
import pytest
|
|
|
|
from app.modules.subscriptions.constants import SubscriptionScheduleType
|
|
from app.modules.subscriptions.services.schedule import (
|
|
ScheduleParseError,
|
|
is_in_quiet_hours,
|
|
next_occurrence,
|
|
next_quiet_end,
|
|
parse_schedule,
|
|
)
|
|
|
|
|
|
NOW = datetime(2026, 7, 26, 0, 0, tzinfo=UTC)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("expression", "schedule_type", "next_run_at"),
|
|
[
|
|
("今天 09:00", SubscriptionScheduleType.ONCE, datetime(2026, 7, 26, 1, 0)),
|
|
("明天 9点半", SubscriptionScheduleType.ONCE, datetime(2026, 7, 27, 1, 30)),
|
|
(
|
|
"2026-08-01 10:15",
|
|
SubscriptionScheduleType.ONCE,
|
|
datetime(2026, 8, 1, 2, 15),
|
|
),
|
|
("每天 09:00", SubscriptionScheduleType.DAILY, datetime(2026, 7, 26, 1, 0)),
|
|
("工作日 09:00", SubscriptionScheduleType.WEEKDAY, datetime(2026, 7, 27, 1, 0)),
|
|
("每周一 09:00", SubscriptionScheduleType.WEEKLY, datetime(2026, 7, 27, 1, 0)),
|
|
("每月26号 09:00", SubscriptionScheduleType.MONTHLY, datetime(2026, 7, 26, 1, 0)),
|
|
(
|
|
"每隔 2 小时",
|
|
SubscriptionScheduleType.INTERVAL,
|
|
datetime(2026, 7, 26, 2, 0),
|
|
),
|
|
],
|
|
)
|
|
def test_parse_supported_chinese_schedules(
|
|
expression: str,
|
|
schedule_type: str,
|
|
next_run_at: datetime,
|
|
) -> None:
|
|
result = parse_schedule(expression, "Asia/Shanghai", now=NOW)
|
|
|
|
assert result.schedule_type == schedule_type
|
|
assert result.next_run_at == next_run_at
|
|
assert result.timezone == "Asia/Shanghai"
|
|
|
|
|
|
def test_monthly_schedule_skips_months_without_the_requested_day() -> None:
|
|
result = parse_schedule(
|
|
"每月31号 09:00",
|
|
"Asia/Shanghai",
|
|
now=datetime(2026, 4, 30, 2, 0, tzinfo=UTC),
|
|
)
|
|
|
|
assert result.next_run_at == datetime(2026, 5, 31, 1, 0)
|
|
assert (
|
|
next_occurrence(
|
|
result.schedule_type,
|
|
result.schedule_config,
|
|
result.timezone,
|
|
after=datetime(2026, 5, 31, 1, 0),
|
|
)
|
|
== datetime(2026, 7, 31, 1, 0)
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"expression",
|
|
[
|
|
"每隔 14 分钟",
|
|
"有空时提醒我",
|
|
"今天 07:59",
|
|
"2026-02-30 09:00",
|
|
"每天 25:00",
|
|
],
|
|
)
|
|
def test_invalid_or_ambiguous_schedule_is_rejected(expression: str) -> None:
|
|
with pytest.raises(ScheduleParseError):
|
|
parse_schedule(expression, "Asia/Shanghai", now=NOW)
|
|
|
|
|
|
def test_invalid_iana_timezone_is_rejected() -> None:
|
|
with pytest.raises(ScheduleParseError, match="Invalid IANA timezone"):
|
|
parse_schedule("每天 09:00", "Mars/Olympus", now=NOW)
|
|
|
|
|
|
def test_quiet_hours_support_cross_midnight_windows() -> None:
|
|
current = datetime(2026, 7, 26, 15, 30, tzinfo=UTC)
|
|
|
|
assert is_in_quiet_hours(
|
|
current,
|
|
"Asia/Shanghai",
|
|
time(22, 0),
|
|
time(7, 0),
|
|
)
|
|
assert next_quiet_end(
|
|
current,
|
|
"Asia/Shanghai",
|
|
time(22, 0),
|
|
time(7, 0),
|
|
) == datetime(2026, 7, 26, 23, 0)
|
|
|