140 lines
4.9 KiB
Python
140 lines
4.9 KiB
Python
"""任务调度器测试：执行日志记录、Cron 校验与执行策略防重叠。"""
|
||
import types
|
||
|
||
import pytest
|
||
|
||
from modules.job import scheduler
|
||
from modules.job_log import service as job_log_service
|
||
|
||
|
||
def make_job(**overrides):
|
||
"""构造一个伪任务快照（无 DB），与调度器要求的 JobSnapshot 对齐。
|
||
|
||
默认 URL 用公网 IP 字面量（8.8.8.8）：无需 DNS 即能通过 getaddrinfo，
|
||
且 is_global=True，符合调度器防 SSRF 的 URL 校验（不再使用 127.0.0.1）。
|
||
"""
|
||
defaults = dict(
|
||
id=1, job_name="测试任务", job_group="系统任务", job_trigger="cron",
|
||
cron_expression="*/5 * * * *", url="http://8.8.8.8/api/v1/health",
|
||
execute_policy=1, status=1,
|
||
)
|
||
defaults.update(overrides)
|
||
return scheduler._snapshot_job(types.SimpleNamespace(**defaults))
|
||
|
||
|
||
class FakeResponse:
|
||
"""支持 with 语法的伪 HTTP 响应。"""
|
||
|
||
status = 200
|
||
|
||
def __enter__(self):
|
||
return self
|
||
|
||
def __exit__(self, *exc):
|
||
return False
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def _clean_running():
|
||
"""每个用例前后清空运行中集合，避免相互污染。"""
|
||
scheduler._running.clear()
|
||
yield
|
||
scheduler._running.clear()
|
||
|
||
|
||
def _capture(monkeypatch):
|
||
"""打桩 record_job_log，返回记录列表。"""
|
||
records = []
|
||
monkeypatch.setattr(job_log_service, "record_job_log", lambda **kw: records.append(kw))
|
||
return records
|
||
|
||
|
||
def test_execute_job_success_records_log(monkeypatch):
|
||
records = _capture(monkeypatch)
|
||
# execute_job 经 _no_redirect_opener.open 发起请求（禁重定向，防 SSRF 302 跳内网）
|
||
monkeypatch.setattr(scheduler._no_redirect_opener, "open", lambda url, timeout=None: FakeResponse())
|
||
|
||
scheduler.execute_job(make_job())
|
||
|
||
assert len(records) == 1
|
||
assert records[0]["status"] == 0
|
||
assert records[0]["job_name"] == "测试任务"
|
||
assert records[0]["job_message"] == "任务触发成功"
|
||
assert records[0]["consume_time"] >= 0
|
||
|
||
|
||
def test_execute_job_failure_records_exception(monkeypatch):
|
||
records = _capture(monkeypatch)
|
||
|
||
def _boom(url, timeout=None):
|
||
raise ConnectionError("connect timeout")
|
||
|
||
monkeypatch.setattr(scheduler._no_redirect_opener, "open", _boom)
|
||
|
||
scheduler.execute_job(make_job())
|
||
|
||
assert len(records) == 1
|
||
assert records[0]["status"] == 1
|
||
assert "connect timeout" in records[0]["exception_info"]
|
||
assert records[0]["job_message"] == "任务触发失败"
|
||
|
||
|
||
def test_overlap_policy_skip_while_running(monkeypatch):
|
||
"""策略 3-放弃执行：任务执行中时跳过本次触发，不调用任务 URL。"""
|
||
records = _capture(monkeypatch)
|
||
calls = {"urlopen": 0}
|
||
|
||
def _urlopen(url, timeout=None):
|
||
calls["urlopen"] += 1
|
||
return FakeResponse()
|
||
|
||
monkeypatch.setattr(scheduler._no_redirect_opener, "open", _urlopen)
|
||
|
||
job = make_job(execute_policy=3)
|
||
scheduler._running.add(job.id)
|
||
scheduler.execute_job(job)
|
||
|
||
assert calls["urlopen"] == 0
|
||
assert len(records) == 1
|
||
assert records[0]["status"] == 0
|
||
assert "放弃执行" in records[0]["job_message"]
|
||
# 跳过不等于释放运行标记：原执行仍占用，任务应保留在运行集合
|
||
assert job.id in scheduler._running
|
||
|
||
|
||
def test_overlap_policy_immediate_runs_in_parallel(monkeypatch):
|
||
"""策略 1-立即执行：即使任务执行中仍再次执行（并行）。"""
|
||
records = _capture(monkeypatch)
|
||
monkeypatch.setattr(scheduler._no_redirect_opener, "open", lambda url, timeout=None: FakeResponse())
|
||
|
||
job = make_job(execute_policy=1)
|
||
scheduler._running.add(job.id)
|
||
scheduler.execute_job(job)
|
||
|
||
assert len(records) == 1
|
||
assert records[0]["status"] == 0
|
||
assert job.id not in scheduler._running # 执行结束从运行集合移除
|
||
|
||
|
||
def test_validate_cron_expression():
|
||
assert scheduler.validate_cron_expression("*/5 * * * *") is None
|
||
assert scheduler.validate_cron_expression("") is not None
|
||
assert scheduler.validate_cron_expression("not-a-cron") is not None
|
||
|
||
|
||
def test_validate_job_url_blocks_ssrf():
|
||
"""防 SSRF：拒绝非 http/https 协议与内网/回环/云元数据地址。"""
|
||
assert scheduler.validate_job_url("http://8.8.8.8/ping") is True
|
||
assert scheduler.validate_job_url("http://127.0.0.1:8041/x") is False
|
||
assert scheduler.validate_job_url("http://192.168.1.1/x") is False
|
||
assert scheduler.validate_job_url("http://169.254.169.254/latest/meta-data/") is False
|
||
assert scheduler.validate_job_url("file:///etc/passwd") is False
|
||
assert scheduler.validate_job_url("ftp://example.com/x") is False
|
||
|
||
|
||
def test_no_redirect_handler_blocks_redirect():
|
||
"""禁重定向：公网 URL 302 跳内网的 SSRF 通道被切断。"""
|
||
handler = scheduler._NoRedirectHandler()
|
||
# redirect_request 返回 None -> urlopen 抛 HTTPError，不会继续请求跳转目标
|
||
assert handler.redirect_request(None, None, 302, "", None, "http://169.254.169.254/") is None
|