166 lines
4.4 KiB
Python
166 lines
4.4 KiB
Python
"""调度器模块测试"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
|
|
import pytest
|
|
|
|
from minenasai.scheduler.cron import CronJob, CronParser, CronScheduler, JobStatus
|
|
|
|
|
|
class TestCronParser:
|
|
"""Cron 解析器测试"""
|
|
|
|
def test_parse_all_stars(self):
|
|
"""测试全星号表达式"""
|
|
result = CronParser.parse("* * * * *")
|
|
|
|
assert len(result["minute"]) == 60
|
|
assert len(result["hour"]) == 24
|
|
assert len(result["day"]) == 31
|
|
assert len(result["month"]) == 12
|
|
assert len(result["weekday"]) == 7
|
|
|
|
def test_parse_specific_values(self):
|
|
"""测试具体值"""
|
|
result = CronParser.parse("30 8 * * *")
|
|
|
|
assert result["minute"] == {30}
|
|
assert result["hour"] == {8}
|
|
|
|
def test_parse_range(self):
|
|
"""测试范围"""
|
|
result = CronParser.parse("0 9-17 * * *")
|
|
|
|
assert result["hour"] == {9, 10, 11, 12, 13, 14, 15, 16, 17}
|
|
|
|
def test_parse_step(self):
|
|
"""测试步进"""
|
|
result = CronParser.parse("*/15 * * * *")
|
|
|
|
assert result["minute"] == {0, 15, 30, 45}
|
|
|
|
def test_parse_list(self):
|
|
"""测试列表"""
|
|
result = CronParser.parse("0 8,12,18 * * *")
|
|
|
|
assert result["hour"] == {8, 12, 18}
|
|
|
|
def test_parse_preset_daily(self):
|
|
"""测试预定义表达式 @daily"""
|
|
result = CronParser.parse("@daily")
|
|
|
|
assert result["minute"] == {0}
|
|
assert result["hour"] == {0}
|
|
|
|
def test_parse_preset_hourly(self):
|
|
"""测试预定义表达式 @hourly"""
|
|
result = CronParser.parse("@hourly")
|
|
|
|
assert result["minute"] == {0}
|
|
assert len(result["hour"]) == 24
|
|
|
|
def test_parse_invalid_expression(self):
|
|
"""测试无效表达式"""
|
|
with pytest.raises(ValueError, match="无效的 cron"):
|
|
CronParser.parse("invalid")
|
|
|
|
def test_get_next_run(self):
|
|
"""测试计算下次运行时间"""
|
|
# 每小时整点
|
|
next_run = CronParser.get_next_run(
|
|
"0 * * * *",
|
|
after=datetime(2026, 1, 1, 10, 30)
|
|
)
|
|
|
|
assert next_run.minute == 0
|
|
assert next_run.hour == 11
|
|
|
|
|
|
class TestCronJob:
|
|
"""CronJob 测试"""
|
|
|
|
def test_job_creation(self):
|
|
"""测试创建任务"""
|
|
job = CronJob(
|
|
id="test-job",
|
|
name="测试任务",
|
|
schedule="*/5 * * * *",
|
|
task="测试",
|
|
)
|
|
|
|
assert job.id == "test-job"
|
|
assert job.enabled is True
|
|
assert job.last_status == JobStatus.PENDING
|
|
|
|
|
|
class TestCronScheduler:
|
|
"""CronScheduler 测试"""
|
|
|
|
def setup_method(self):
|
|
"""初始化"""
|
|
self.scheduler = CronScheduler()
|
|
|
|
def test_add_job(self):
|
|
"""测试添加任务"""
|
|
async def task():
|
|
pass
|
|
|
|
job = self.scheduler.add_job(
|
|
job_id="test-1",
|
|
name="测试任务",
|
|
schedule="*/5 * * * *",
|
|
callback=task,
|
|
)
|
|
|
|
assert job.id == "test-1"
|
|
assert job.next_run is not None
|
|
|
|
def test_remove_job(self):
|
|
"""测试移除任务"""
|
|
async def task():
|
|
pass
|
|
|
|
self.scheduler.add_job("test-1", "测试", "* * * * *", task)
|
|
|
|
assert self.scheduler.remove_job("test-1") is True
|
|
assert self.scheduler.get_job("test-1") is None
|
|
|
|
def test_enable_disable_job(self):
|
|
"""测试启用/禁用任务"""
|
|
async def task():
|
|
pass
|
|
|
|
self.scheduler.add_job("test-1", "测试", "* * * * *", task)
|
|
|
|
assert self.scheduler.disable_job("test-1") is True
|
|
job = self.scheduler.get_job("test-1")
|
|
assert job.enabled is False
|
|
assert job.last_status == JobStatus.DISABLED
|
|
|
|
assert self.scheduler.enable_job("test-1") is True
|
|
assert job.enabled is True
|
|
|
|
def test_list_jobs(self):
|
|
"""测试列出任务"""
|
|
async def task():
|
|
pass
|
|
|
|
self.scheduler.add_job("test-1", "任务1", "* * * * *", task)
|
|
self.scheduler.add_job("test-2", "任务2", "*/5 * * * *", task)
|
|
|
|
jobs = self.scheduler.list_jobs()
|
|
assert len(jobs) == 2
|
|
|
|
def test_get_stats(self):
|
|
"""测试获取统计"""
|
|
async def task():
|
|
pass
|
|
|
|
self.scheduler.add_job("test-1", "任务1", "* * * * *", task)
|
|
|
|
stats = self.scheduler.get_stats()
|
|
assert stats["total_jobs"] == 1
|
|
assert stats["enabled_jobs"] == 1
|