Files
dl-organizer/tests/test_cli_review_plan.py
T

89 lines
2.9 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"',
]
)
)
def test_review_plan_generates_csv_and_summary(tmp_path):
"""review-plan should export flagged operations and summary counters."""
config_path = tmp_path / "config.yaml"
_write_config(config_path, tmp_path / "library")
plan_path = tmp_path / "plan.json"
plan_data = {
"plan_id": "test-plan",
"created_at": "2026-02-13T00:00:00+00:00",
"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,
},
],
"summary": {"total": 2, "move": 1, "rename": 0, "quarantine": 0, "no-op": 1},
"summary_by_reason": {},
"human_summary": "",
"metadata": {},
}
plan_path.write_text(json.dumps(plan_data), encoding="utf-8")
output_csv = tmp_path / "review.csv"
runner = CliRunner()
result = runner.invoke(
main,
[
"--config", str(config_path),
"review-plan",
"--input", str(plan_path),
"--output", str(output_csv),
"--season-threshold", "20",
"--episode-threshold", "40",
],
)
assert result.exit_code == 0
assert "Plan review summary:" in result.output
assert "High-risk operations: 2" 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)