windyboyandCursor
5f0b531269
Make review-plan easier to act on: Chinese risk labels, relative paths, verdict/next-step footer, optional identity/analysis enrichment, grouping, spot-check sampling, and structure preview. Extend the TUI with filters, duplicate-group reject, and quality context. Persist review_context on plan operations and add --require-review for confirmed execute. Co-authored-by: Cursor <cursoragent@cursor.com>
125 lines
4.0 KiB
Python
125 lines
4.0 KiB
Python
"""Tests for plan review helpers."""
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
from vlm.models import ExecutionPlan, FileOperation
|
|
from vlm.plan_review import (
|
|
build_identity_lookup,
|
|
check_review_requirements,
|
|
enrich_review_rows,
|
|
prepare_review_rows,
|
|
review_plan,
|
|
)
|
|
from vlm.plan_render import render_review_verdict
|
|
from vlm.plan_structure_preview import build_structure_preview_lines
|
|
from vlm.planner import load_plan, save_plan
|
|
from vlm.utils import utc_now
|
|
|
|
|
|
def _minimal_plan(operations: list[FileOperation]) -> ExecutionPlan:
|
|
return ExecutionPlan(
|
|
plan_id="t",
|
|
created_at=utc_now(),
|
|
operations=operations,
|
|
summary={"total": len(operations), "move": 1, "rename": 0, "quarantine": 0, "no-op": 0},
|
|
)
|
|
|
|
|
|
def test_render_review_verdict_blocked_and_ok():
|
|
assert "BLOCKED" in render_review_verdict({"high_risk_operations": 2})
|
|
assert "OK TO DRY-RUN" in render_review_verdict({"high_risk_operations": 0})
|
|
|
|
|
|
def test_build_identity_lookup_and_enrich(tmp_path):
|
|
identities = {
|
|
"vlm_schema_version": "1.0",
|
|
"movies": [],
|
|
"series": [
|
|
{
|
|
"path": str(tmp_path / "dl/Show.S01E01.mkv"),
|
|
"filename": "Show.S01E01.mkv",
|
|
"category": "series",
|
|
"title": "Show",
|
|
"season": 1,
|
|
"episodes": [1],
|
|
"confidence": 1.0,
|
|
"needs_review": False,
|
|
"video_metadata": {"resolution": "1080p", "size_bytes": 1000},
|
|
}
|
|
],
|
|
}
|
|
lookup = build_identity_lookup(identities)
|
|
op = FileOperation(
|
|
operation_type="move",
|
|
source_path=tmp_path / "dl/Show.S01E01.mkv",
|
|
destination_path=tmp_path / "lib/series/Show/Season 01/S01E01.mkv",
|
|
reason="organize",
|
|
has_conflict=False,
|
|
review_context={"title": "Show", "category": "series", "season": 1, "episode": 1},
|
|
)
|
|
plan = _minimal_plan([op])
|
|
rows = [
|
|
{
|
|
"index": "1",
|
|
"operation_type": "move",
|
|
"risk_flags": "",
|
|
"source_path": str(op.source_path),
|
|
"destination_path": str(op.destination_path),
|
|
"reason": op.reason,
|
|
}
|
|
]
|
|
enriched = enrich_review_rows(rows, plan, identity_lookup=lookup, library_root=tmp_path / "lib")
|
|
assert enriched[0]["title"] == "Show"
|
|
assert enriched[0]["quality_hint"]
|
|
assert enriched[0]["source_name"] == "Show.S01E01.mkv"
|
|
|
|
|
|
def test_check_review_requirements_missing_csv(tmp_path):
|
|
plan_path = tmp_path / "plan.json"
|
|
op = FileOperation(
|
|
operation_type="no-op",
|
|
source_path=tmp_path / "Series.S20E01.mkv",
|
|
destination_path=None,
|
|
reason="Series needs manual review (season exceeds configured threshold)",
|
|
has_conflict=False,
|
|
)
|
|
plan = _minimal_plan([op])
|
|
save_plan(plan, plan_path)
|
|
loaded = load_plan(plan_path)
|
|
errors = check_review_requirements(loaded, plan_path, tmp_path / "missing.csv")
|
|
assert len(errors) == 1
|
|
assert "not found" in errors[0]
|
|
|
|
|
|
def test_structure_preview_lines(tmp_path):
|
|
lib = tmp_path / "library"
|
|
op = FileOperation(
|
|
operation_type="move",
|
|
source_path=tmp_path / "a.mkv",
|
|
destination_path=lib / "series/Show/Season 01/S01E01.mkv",
|
|
reason="x",
|
|
has_conflict=False,
|
|
review_context={"title": "Show"},
|
|
)
|
|
lines = build_structure_preview_lines(_minimal_plan([op]), lib)
|
|
assert any("Show" in line for line in lines)
|
|
|
|
|
|
def test_prepare_review_rows_sample_safe(tmp_path):
|
|
ops = [
|
|
FileOperation(
|
|
operation_type="move",
|
|
source_path=tmp_path / f"f{i}.mkv",
|
|
destination_path=tmp_path / f"lib/f{i}.mkv",
|
|
reason="organize",
|
|
has_conflict=False,
|
|
)
|
|
for i in range(5)
|
|
]
|
|
plan = _minimal_plan(ops)
|
|
rows, counters = prepare_review_rows(plan, sample_safe=2)
|
|
assert counters["high_risk_operations"] == 0
|
|
assert len(rows) == 2
|
|
assert all(r["risk_flags"] == "spot_check" for r in rows)
|