Files
dl-organizer/tests/test_cli_review_plan.py
79797644e1 chore: trim dead code, modularize CLI, and archive stale docs
Extract review-plan, report, quarantine, state, and config handlers into
commands/ with shared cli_helpers; remove unused exceptions and duplicate
plan summary wrappers. Archive superseded review markdown, sync docs to
517-test baseline, and fix empty series titles when only a quality tag remains.

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
Co-authored-by: Cursor <cursoragent@cursor.com>
2026-05-21 10:36:03 +08:00

348 lines
11 KiB
Python

"""Tests for `vlm review-plan` CLI command."""
import csv
import json
from pathlib import Path
from click.testing import CliRunner
from vlm.cli import main
def _write_config(path: Path, library_root: Path) -> None:
path.write_text(
"\n".join(
[
f"library_root: {library_root}",
"video_extensions:",
" - .mkv",
"templates:",
' movie_dir: "movie/{title} ({year})/"',
' series_dir: "series/{title}/Season {season:02d}/"',
' movie_filename: "{title} ({year}){ext}"',
' series_filename: "S{season:02d}E{episode:02d}{ext}"',
'quarantine_dir: ".quarantine"',
'log_level: "INFO"',
]
),
encoding="utf-8",
)
def _write_plan(path: Path, operations: list[dict], summary: dict, human_summary: str = "") -> None:
plan_data = {
"vlm_schema_version": "1.0",
"plan_id": "test-plan",
"created_at": "2026-02-13T00:00:00+00:00",
"operations": operations,
"summary": summary,
"summary_by_reason": {},
"human_summary": human_summary,
"metadata": {},
}
path.write_text(json.dumps(plan_data), encoding="utf-8")
def _invoke_review_plan(runner: CliRunner, config_path: Path, args: list[str]):
return runner.invoke(main, ["--config", str(config_path), "review-plan", *args])
def test_review_plan_generates_csv_summary_and_preview(tmp_path):
"""review-plan should export CSV, show summary, and print preview content."""
config_path = tmp_path / "config.yaml"
_write_config(config_path, tmp_path / "library")
plan_path = tmp_path / "plan.json"
operations = [
{
"operation_type": "move",
"source_path": str(tmp_path / "Show.Sample.S01E01.mkv"),
"destination_path": str(tmp_path / "library/series/Show/Season 01/S01E01.mkv"),
"reason": "Organize series: Show S01E01",
"has_conflict": False,
"conflict_reason": None,
},
{
"operation_type": "no-op",
"source_path": str(tmp_path / "Show.S20E50.mkv"),
"destination_path": None,
"reason": "Series needs manual review (season exceeds configured threshold)",
"has_conflict": False,
"conflict_reason": None,
},
]
_write_plan(
plan_path,
operations=operations,
summary={"total": 2, "move": 1, "rename": 0, "quarantine": 0, "no-op": 1},
human_summary="这是测试计划摘要",
)
output_csv = tmp_path / "review.csv"
runner = CliRunner()
result = _invoke_review_plan(
runner,
config_path,
[
"--input",
str(plan_path),
"--output",
str(output_csv),
"--season-threshold",
"20",
"--episode-threshold",
"40",
],
)
assert result.exit_code == 0
assert "Plan overview:" in result.output
assert "这是测试计划摘要" in result.output
assert "Plan review summary:" in result.output
assert "High-risk operations: 1" in result.output
assert "High-risk operations preview:" in result.output
assert "source:" in result.output
assert "destination:" in result.output
assert "reason:" in result.output
assert "Verdict: BLOCKED" in result.output
assert "需人工判断" in result.output or "样片路径" in result.output
assert "Next steps:" in result.output
assert "apply-review" in result.output
assert output_csv.exists()
with open(output_csv, "r", encoding="utf-8", newline="") as f:
rows = list(csv.DictReader(f))
assert len(rows) == 2
assert any("sample_source" in row["risk_flags"] for row in rows)
assert any("manual_review" in row["risk_flags"] for row in rows)
def test_review_plan_preview_limit_controls_console_output(tmp_path):
"""--preview-limit should cap displayed high-risk operations and report remaining count."""
config_path = tmp_path / "config.yaml"
_write_config(config_path, tmp_path / "library")
plan_path = tmp_path / "plan.json"
operations = [
{
"operation_type": "no-op",
"source_path": str(tmp_path / "Series.S20E41.mkv"),
"destination_path": None,
"reason": "Series needs manual review (season exceeds configured threshold)",
"has_conflict": False,
"conflict_reason": None,
},
{
"operation_type": "no-op",
"source_path": str(tmp_path / "Series.S20E42.mkv"),
"destination_path": None,
"reason": "Series needs manual review (season exceeds configured threshold)",
"has_conflict": False,
"conflict_reason": None,
},
{
"operation_type": "no-op",
"source_path": str(tmp_path / "Series.S20E43.mkv"),
"destination_path": None,
"reason": "Series needs manual review (season exceeds configured threshold)",
"has_conflict": False,
"conflict_reason": None,
},
]
_write_plan(
plan_path,
operations=operations,
summary={"total": 3, "move": 0, "rename": 0, "quarantine": 0, "no-op": 3},
)
output_csv = tmp_path / "review.csv"
runner = CliRunner()
result = _invoke_review_plan(
runner,
config_path,
[
"--input",
str(plan_path),
"--output",
str(output_csv),
"--preview-limit",
"1",
],
)
assert result.exit_code == 0
assert " - [1] " in result.output
assert " - [2] " not in result.output
assert " - [3] " not in result.output
assert "... and 2 more high-risk operations" in result.output
def test_review_plan_show_all_overrides_preview_limit(tmp_path):
"""--show-all should display all high-risk operations regardless of --preview-limit."""
config_path = tmp_path / "config.yaml"
_write_config(config_path, tmp_path / "library")
plan_path = tmp_path / "plan.json"
operations = [
{
"operation_type": "no-op",
"source_path": str(tmp_path / "Series.S20E51.mkv"),
"destination_path": None,
"reason": "Series needs manual review (season exceeds configured threshold)",
"has_conflict": False,
"conflict_reason": None,
},
{
"operation_type": "no-op",
"source_path": str(tmp_path / "Series.S20E52.mkv"),
"destination_path": None,
"reason": "Series needs manual review (season exceeds configured threshold)",
"has_conflict": False,
"conflict_reason": None,
},
{
"operation_type": "no-op",
"source_path": str(tmp_path / "Series.S20E53.mkv"),
"destination_path": None,
"reason": "Series needs manual review (season exceeds configured threshold)",
"has_conflict": False,
"conflict_reason": None,
},
]
_write_plan(
plan_path,
operations=operations,
summary={"total": 3, "move": 0, "rename": 0, "quarantine": 0, "no-op": 3},
)
output_csv = tmp_path / "review.csv"
runner = CliRunner()
result = _invoke_review_plan(
runner,
config_path,
[
"--input",
str(plan_path),
"--output",
str(output_csv),
"--preview-limit",
"1",
"--show-all",
],
)
assert result.exit_code == 0
assert " - [1] " in result.output
assert " - [2] " in result.output
assert " - [3] " in result.output
assert "more high-risk operations" not in result.output
def test_review_plan_tui_requires_interactive_terminal(tmp_path):
"""--tui should fail when stdin/stdout are not TTY (e.g. CliRunner)."""
config_path = tmp_path / "config.yaml"
_write_config(config_path, tmp_path / "library")
plan_path = tmp_path / "plan.json"
operations = [
{
"operation_type": "move",
"source_path": str(tmp_path / "a.mkv"),
"destination_path": str(tmp_path / "b.mkv"),
"reason": "organize",
"has_conflict": False,
"conflict_reason": None,
},
]
_write_plan(
plan_path,
operations=operations,
summary={"total": 1, "move": 1, "rename": 0, "quarantine": 0, "no-op": 0},
)
output_csv = tmp_path / "review.csv"
runner = CliRunner()
result = _invoke_review_plan(
runner,
config_path,
["--input", str(plan_path), "--output", str(output_csv), "--tui"],
)
assert result.exit_code == 1
assert "TTY" in result.output or "interactive terminal" in result.output
def test_review_plan_tui_stubbed_success(tmp_path, monkeypatch):
"""With TTY check stubbed and run_plan_review_tui stubbed, CLI should succeed."""
config_path = tmp_path / "config.yaml"
_write_config(config_path, tmp_path / "library")
plan_path = tmp_path / "plan.json"
operations = [
{
"operation_type": "move",
"source_path": str(tmp_path / "a.mkv"),
"destination_path": str(tmp_path / "b.mkv"),
"reason": "organize",
"has_conflict": False,
"conflict_reason": None,
},
]
_write_plan(
plan_path,
operations=operations,
summary={"total": 1, "move": 1, "rename": 0, "quarantine": 0, "no-op": 0},
)
monkeypatch.setattr("vlm.commands.review_plan.review_plan_tui_streams_ok", lambda: True)
monkeypatch.setattr("vlm.review_tui.run_plan_review_tui", lambda ctx: 0)
output_csv = tmp_path / "review.csv"
runner = CliRunner()
result = _invoke_review_plan(
runner,
config_path,
["--input", str(plan_path), "--output", str(output_csv), "--tui"],
)
assert result.exit_code == 0
assert "Saved manual review CSV" in result.output
assert "Plan overview:" not in result.output
def test_review_plan_tui_stubbed_abort(tmp_path, monkeypatch):
"""TUI return code 1 should surface as CLI failure."""
config_path = tmp_path / "config.yaml"
_write_config(config_path, tmp_path / "library")
plan_path = tmp_path / "plan.json"
operations = [
{
"operation_type": "move",
"source_path": str(tmp_path / "a.mkv"),
"destination_path": str(tmp_path / "b.mkv"),
"reason": "organize",
"has_conflict": False,
"conflict_reason": None,
},
]
_write_plan(
plan_path,
operations=operations,
summary={"total": 1, "move": 1, "rename": 0, "quarantine": 0, "no-op": 0},
)
monkeypatch.setattr("vlm.commands.review_plan.review_plan_tui_streams_ok", lambda: True)
monkeypatch.setattr("vlm.review_tui.run_plan_review_tui", lambda ctx: 1)
output_csv = tmp_path / "review.csv"
runner = CliRunner()
result = _invoke_review_plan(
runner,
config_path,
["--input", str(plan_path), "--output", str(output_csv), "--tui"],
)
assert result.exit_code == 1
assert "aborted" in result.output.lower()