50 lines
1.6 KiB
Python
50 lines
1.6 KiB
Python
"""
|
||
定时任务服务测试（F4 调度器入参修复）
|
||
=============================================================
|
||
|
||
验证 JobService 与调度器的联动契约：
|
||
- add_job_to_scheduler 应收到 dict（驼峰键，含 cronExpression/id/status），
|
||
而非 ORM 实例（原实现传 ORM 实例 → AttributeError 被吞 → 任务永不注册）
|
||
"""
|
||
|
||
from src.core.database import db
|
||
from src.modules.job.service import JobService
|
||
|
||
|
||
def _job_payload():
|
||
"""构造合法任务表单数据（isSync=0 异步：回归验证 DataRequired→Optional 修复）"""
|
||
return {
|
||
"jobName": "test_job",
|
||
"status": 1,
|
||
"executePolicy": 1,
|
||
"isSync": 0,
|
||
"cronExpression": "0 2 * * *",
|
||
"url": "http://localhost/api/v1/notify",
|
||
"note": "测试",
|
||
}
|
||
|
||
|
||
class TestJobSchedulerDict:
|
||
"""调度器入参应为驼峰 dict"""
|
||
|
||
def test_add_passes_dict_to_scheduler(self, app, monkeypatch):
|
||
captured = {}
|
||
|
||
def fake_add(job_dict):
|
||
captured["arg"] = job_dict
|
||
return True
|
||
|
||
monkeypatch.setattr("src.modules.job.service.add_job_to_scheduler", fake_add)
|
||
|
||
with app.app_context():
|
||
JobService.add(_job_payload())
|
||
db.session.commit()
|
||
|
||
assert "arg" in captured
|
||
arg = captured["arg"]
|
||
# 必须是 dict 且含调度器所需驼峰键（原 ORM 实例无 .get → AttributeError）
|
||
assert isinstance(arg, dict)
|
||
assert arg["cronExpression"] == "0 2 * * *"
|
||
assert arg["status"] == 1
|
||
assert arg["id"] is not None
|