133 lines
4.4 KiB
Python
133 lines
4.4 KiB
Python
"""M7 定时任务测试。
|
|
|
|
使用 mock subprocess.run,不依赖真实脚本执行。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import subprocess
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from scheduler import PipelineResult, StepResult, run_pipeline, run_step
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# fixtures
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
def _mock_proc(returncode: int = 0, stderr: str = "") -> MagicMock:
|
|
p = MagicMock()
|
|
p.returncode = returncode
|
|
p.stderr = stderr
|
|
p.stdout = "ok"
|
|
return p
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# StepResult 模型
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
def test_step_result_defaults() -> None:
|
|
sr = StepResult(name="crawler", success=True, elapsed_sec=12.3)
|
|
assert sr.name == "crawler"
|
|
assert sr.success is True
|
|
assert sr.elapsed_sec == 12.3
|
|
assert sr.exit_code is None
|
|
|
|
|
|
def test_pipeline_result_all_success() -> None:
|
|
pr = PipelineResult(steps=[
|
|
StepResult(name="crawler", success=True, elapsed_sec=1),
|
|
StepResult(name="extractor", success=True, elapsed_sec=2),
|
|
])
|
|
assert pr.all_success is True
|
|
|
|
|
|
def test_pipeline_result_partial_failure() -> None:
|
|
pr = PipelineResult(steps=[
|
|
StepResult(name="crawler", success=True, elapsed_sec=1),
|
|
StepResult(name="extractor", success=False, elapsed_sec=0),
|
|
])
|
|
assert pr.all_success is False
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# run_step
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
def test_run_step_success() -> None:
|
|
with patch("subprocess.run", return_value=_mock_proc(returncode=0,
|
|
stderr="INFO | 完成: 成功 20/20")):
|
|
sr = run_step("crawler", "20260616")
|
|
assert sr.success is True
|
|
assert sr.exit_code == 0
|
|
|
|
|
|
def test_run_step_failure() -> None:
|
|
with patch("subprocess.run", return_value=_mock_proc(returncode=1)):
|
|
sr = run_step("extractor", "20260616")
|
|
assert sr.success is False
|
|
assert sr.exit_code == 1
|
|
assert "rc=1" in sr.tail_msg
|
|
|
|
|
|
def test_run_step_timeout() -> None:
|
|
with patch("subprocess.run", side_effect=subprocess.TimeoutExpired(cmd=["uv"], timeout=10)):
|
|
sr = run_step("llm", "20260616")
|
|
assert sr.success is False
|
|
assert "超时" in sr.tail_msg
|
|
|
|
|
|
def test_run_step_exception() -> None:
|
|
with patch("subprocess.run", side_effect=OSError("磁盘满")):
|
|
sr = run_step("embedding", "20260616")
|
|
assert sr.success is False
|
|
assert "磁盘满" in sr.tail_msg
|
|
|
|
|
|
def test_run_step_unknown_name() -> None:
|
|
sr = run_step("nonexistent", "20260616")
|
|
assert sr.success is False
|
|
assert "未知" in sr.tail_msg
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# run_pipeline
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
def test_run_pipeline_all_success() -> None:
|
|
with (
|
|
patch("subprocess.run", return_value=_mock_proc(returncode=0)),
|
|
patch("scheduler.reporter.generate_report", return_value=Path("/tmp/r.html")),
|
|
):
|
|
result = run_pipeline("20260616", steps=["crawler", "extractor", "dedup"])
|
|
assert len(result.steps) == 3
|
|
assert result.all_success is True
|
|
assert result.started_at is not None
|
|
assert result.finished_at is not None
|
|
|
|
|
|
def test_run_pipeline_continues_on_failure() -> None:
|
|
"""中间步骤失败,后续继续执行(不阻断)。"""
|
|
call_count = {"n": 0}
|
|
|
|
def _side_effect(*args, **kwargs):
|
|
call_count["n"] += 1
|
|
if call_count["n"] == 2: # extractor 失败
|
|
return _mock_proc(returncode=1, stderr="GNE 提取异常")
|
|
return _mock_proc(returncode=0, stderr="ok")
|
|
|
|
with patch("subprocess.run", side_effect=_side_effect):
|
|
result = run_pipeline("20260616", steps=["crawler", "extractor", "dedup", "llm"])
|
|
assert len(result.steps) == 4
|
|
# extractor 失败,但后续仍执行
|
|
assert result.steps[1].success is False
|
|
assert result.steps[2].success is True
|
|
|
|
|
|
def test_run_pipeline_custom_steps() -> None:
|
|
with patch("subprocess.run", return_value=_mock_proc(returncode=0, stderr="ok")):
|
|
result = run_pipeline("20260616", steps=["crawler", "extractor"])
|
|
assert len(result.steps) == 2
|
|
assert result.all_success is True
|