"""Tests for plan review helpers.""" from vlm.models import ExecutionPlan, FileOperation from vlm.plan_render import render_review_verdict from vlm.plan_review import ( build_duplicate_path_maps, build_identity_lookup, check_review_requirements, enrich_review_rows, normalized_path_key, prepare_review_rows, review_plan, save_review_csv, ) from vlm.plan_structure_preview import build_structure_preview_lines from vlm.planner import apply_review_to_plan, 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="quarantine", 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_check_review_requirements_requires_apply_review(tmp_path): plan_path = tmp_path / "plan.json" csv_path = tmp_path / "plan_manual_review.csv" op = FileOperation( operation_type="move", source_path=tmp_path / "a.mkv", destination_path=tmp_path / "lib/a.mkv", reason="Series needs manual review (season exceeds configured threshold)", has_conflict=True, ) plan = _minimal_plan([op]) save_plan(plan, plan_path) rows, _ = review_plan(plan) save_review_csv(rows, csv_path) errors = check_review_requirements(load_plan(plan_path), plan_path, csv_path) assert len(errors) == 1 assert "apply-review" in errors[0] def test_check_review_requirements_passes_after_apply_review(tmp_path): plan_path = tmp_path / "plan.json" csv_path = tmp_path / "plan_manual_review.csv" op = FileOperation( operation_type="move", source_path=tmp_path / "a.mkv", destination_path=tmp_path / "lib/a.mkv", reason="Series needs manual review (season exceeds configured threshold)", has_conflict=True, ) plan = _minimal_plan([op]) save_plan(plan, plan_path) rows, _ = review_plan(plan) for row in rows: row["operation_type"] = "no-op" save_review_csv(rows, csv_path) updated = apply_review_to_plan(load_plan(plan_path), csv_path) save_plan(updated, plan_path) errors = check_review_requirements(load_plan(plan_path), plan_path, csv_path) assert errors == [] def test_planner_noop_manual_review_does_not_block_execute_gate(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) errors = check_review_requirements(load_plan(plan_path), plan_path, tmp_path / "missing.csv") assert errors == [] def test_review_plan_skips_applied_noop_reason(tmp_path): op = FileOperation( operation_type="no-op", source_path=tmp_path / "a.mkv", destination_path=None, reason="Modified via manual review: Series needs manual review (season exceeds configured threshold)", has_conflict=False, ) rows, counters = review_plan(_minimal_plan([op])) assert rows == [] assert counters["high_risk_operations"] == 0 def test_build_duplicate_path_maps_uses_canonical_keys(tmp_path): raw = str(tmp_path / "movie.mkv") analysis = { "duplicates": [ { "identity": {"type": "movie", "title": "X", "year": 2020}, "files": [raw], "quality_comparison": [{"path": raw, "resolution": "1080p"}], } ] } path_to_group, path_to_quality = build_duplicate_path_maps(analysis) key = normalized_path_key(raw) assert path_to_group[key] == "movie:X:2020" assert path_to_quality[key]["resolution"] == "1080p" 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)