1322 lines
43 KiB
Python
1322 lines
43 KiB
Python
"""Unit tests for plan generator."""
|
|
|
|
import json
|
|
import pytest
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
from vlm.config import Config
|
|
from vlm.models import (
|
|
ExecutionPlan,
|
|
FileOperation,
|
|
MovieIdentity,
|
|
SeriesIdentity,
|
|
VideoFile,
|
|
)
|
|
from vlm.planner import generate_plan, load_plan
|
|
|
|
|
|
@pytest.fixture
|
|
def config():
|
|
"""Create a test configuration."""
|
|
return Config(
|
|
library_root=Path("/mnt/nas/videos"),
|
|
video_extensions=[".mp4", ".mkv", ".avi"],
|
|
movie_template="movie/{title} ({year})/",
|
|
series_template="series/{title}/Season {season:02d}/",
|
|
movie_filename_template="{title} ({year}){ext}",
|
|
series_filename_template="S{season:02d}E{episode:02d}{ext}",
|
|
log_level="INFO",
|
|
quarantine_dir=".quarantine"
|
|
)
|
|
|
|
|
|
def test_generate_plan_for_movie_with_year(config):
|
|
"""Test plan generation for a movie with year."""
|
|
video_file = VideoFile(
|
|
path=Path("/mnt/nas/videos/movie/Some.Movie.2020.1080p.mkv"),
|
|
filename="Some.Movie.2020.1080p.mkv",
|
|
size_bytes=1000000,
|
|
modified_timestamp=datetime.now(timezone.utc),
|
|
category="movie"
|
|
)
|
|
|
|
identity = MovieIdentity(
|
|
title="Some Movie",
|
|
year=2020,
|
|
confidence=0.9,
|
|
needs_review=False,
|
|
original_filename="Some.Movie.2020.1080p.mkv"
|
|
)
|
|
|
|
plan = generate_plan([(video_file, identity)], config)
|
|
|
|
assert isinstance(plan, ExecutionPlan)
|
|
assert len(plan.operations) == 1
|
|
|
|
operation = plan.operations[0]
|
|
assert operation.operation_type == "move"
|
|
assert operation.source_path == video_file.path
|
|
assert operation.destination_path == Path("/mnt/nas/videos/movie/Some Movie (2020)/Some Movie (2020).mkv")
|
|
assert not operation.has_conflict
|
|
assert "Some Movie (2020)" in operation.reason
|
|
|
|
|
|
def test_generate_plan_for_movie_without_year(config):
|
|
"""Test plan generation for a movie without year (needs review)."""
|
|
video_file = VideoFile(
|
|
path=Path("/mnt/nas/videos/movie/random_movie.mkv"),
|
|
filename="random_movie.mkv",
|
|
size_bytes=1000000,
|
|
modified_timestamp=datetime.now(timezone.utc),
|
|
category="movie"
|
|
)
|
|
|
|
identity = MovieIdentity(
|
|
title="Random Movie",
|
|
year=None,
|
|
confidence=0.3,
|
|
needs_review=True,
|
|
original_filename="random_movie.mkv"
|
|
)
|
|
|
|
plan = generate_plan([(video_file, identity)], config)
|
|
|
|
assert len(plan.operations) == 1
|
|
operation = plan.operations[0]
|
|
assert operation.operation_type == "no-op"
|
|
assert operation.destination_path is None
|
|
assert "manual review" in operation.reason.lower()
|
|
|
|
|
|
def test_generate_plan_for_series_with_season_and_episode(config):
|
|
"""Test plan generation for a series with season and episode."""
|
|
video_file = VideoFile(
|
|
path=Path("/mnt/nas/videos/series/Show.Name.S01E05.mkv"),
|
|
filename="Show.Name.S01E05.mkv",
|
|
size_bytes=500000,
|
|
modified_timestamp=datetime.now(timezone.utc),
|
|
category="series"
|
|
)
|
|
|
|
identity = SeriesIdentity(
|
|
title="Show Name",
|
|
season=1,
|
|
episodes=[5],
|
|
confidence=0.9,
|
|
needs_review=False,
|
|
original_filename="Show.Name.S01E05.mkv"
|
|
)
|
|
|
|
plan = generate_plan([(video_file, identity)], config)
|
|
|
|
assert len(plan.operations) == 1
|
|
operation = plan.operations[0]
|
|
assert operation.operation_type == "move"
|
|
assert operation.source_path == video_file.path
|
|
assert operation.destination_path == Path("/mnt/nas/videos/series/Show Name/Season 01/S01E05.mkv")
|
|
assert not operation.has_conflict
|
|
|
|
|
|
def test_generate_plan_blocks_series_with_high_season(config):
|
|
"""Series season above threshold should be no-op for manual review."""
|
|
video_file = VideoFile(
|
|
path=Path("/mnt/nas/videos/series/Show.Name.S20E01.mkv"),
|
|
filename="Show.Name.S20E01.mkv",
|
|
size_bytes=500000,
|
|
modified_timestamp=datetime.now(timezone.utc),
|
|
category="series"
|
|
)
|
|
identity = SeriesIdentity(
|
|
title="Show Name",
|
|
season=20,
|
|
episodes=[1],
|
|
confidence=0.9,
|
|
needs_review=False,
|
|
original_filename="Show.Name.S20E01.mkv"
|
|
)
|
|
|
|
plan = generate_plan([(video_file, identity)], config)
|
|
operation = plan.operations[0]
|
|
assert operation.operation_type == "no-op"
|
|
assert "season exceeds configured threshold" in operation.reason
|
|
|
|
|
|
def test_generate_plan_blocks_series_with_high_episode(config):
|
|
"""Series episode above threshold should be no-op for manual review."""
|
|
video_file = VideoFile(
|
|
path=Path("/mnt/nas/videos/series/Show.Name.S01E120.mkv"),
|
|
filename="Show.Name.S01E120.mkv",
|
|
size_bytes=500000,
|
|
modified_timestamp=datetime.now(timezone.utc),
|
|
category="series"
|
|
)
|
|
identity = SeriesIdentity(
|
|
title="Show Name",
|
|
season=1,
|
|
episodes=[120],
|
|
confidence=0.9,
|
|
needs_review=False,
|
|
original_filename="Show.Name.S01E120.mkv"
|
|
)
|
|
|
|
plan = generate_plan([(video_file, identity)], config)
|
|
operation = plan.operations[0]
|
|
assert operation.operation_type == "no-op"
|
|
assert "episode exceeds configured threshold" in operation.reason
|
|
|
|
|
|
def test_generate_plan_sample_is_noop_by_default(config):
|
|
"""Sample files should be excluded from organize actions by default."""
|
|
video_file = VideoFile(
|
|
path=Path("/mnt/nas/videos/series/Show.Name.Sample.S01E01.mkv"),
|
|
filename="Show.Name.Sample.S01E01.mkv",
|
|
size_bytes=500000,
|
|
modified_timestamp=datetime.now(timezone.utc),
|
|
category="series"
|
|
)
|
|
identity = SeriesIdentity(
|
|
title="Show Name",
|
|
season=1,
|
|
episodes=[1],
|
|
confidence=0.9,
|
|
needs_review=False,
|
|
original_filename="Show.Name.Sample.S01E01.mkv"
|
|
)
|
|
|
|
plan = generate_plan([(video_file, identity)], config)
|
|
operation = plan.operations[0]
|
|
assert operation.operation_type == "no-op"
|
|
assert "Sample file excluded" in operation.reason
|
|
|
|
|
|
def test_generate_plan_sample_can_be_included_via_config(config):
|
|
"""When include_sample_files is enabled, sample files can be organized."""
|
|
config.plan_include_sample_files = True
|
|
video_file = VideoFile(
|
|
path=Path("/mnt/nas/videos/series/Show.Name.Sample.S01E01.mkv"),
|
|
filename="Show.Name.Sample.S01E01.mkv",
|
|
size_bytes=500000,
|
|
modified_timestamp=datetime.now(timezone.utc),
|
|
category="series"
|
|
)
|
|
identity = SeriesIdentity(
|
|
title="Show Name",
|
|
season=1,
|
|
episodes=[1],
|
|
confidence=0.9,
|
|
needs_review=False,
|
|
original_filename="Show.Name.Sample.S01E01.mkv"
|
|
)
|
|
|
|
plan = generate_plan([(video_file, identity)], config)
|
|
operation = plan.operations[0]
|
|
assert operation.operation_type == "move"
|
|
|
|
|
|
def test_generate_plan_for_series_without_season(config):
|
|
"""Test plan generation for a series without season (needs review)."""
|
|
video_file = VideoFile(
|
|
path=Path("/mnt/nas/videos/series/ambiguous_show.mkv"),
|
|
filename="ambiguous_show.mkv",
|
|
size_bytes=500000,
|
|
modified_timestamp=datetime.now(timezone.utc),
|
|
category="series"
|
|
)
|
|
|
|
identity = SeriesIdentity(
|
|
title="Ambiguous Show",
|
|
season=None,
|
|
episodes=[],
|
|
confidence=0.3,
|
|
needs_review=True,
|
|
original_filename="ambiguous_show.mkv"
|
|
)
|
|
|
|
plan = generate_plan([(video_file, identity)], config)
|
|
|
|
assert len(plan.operations) == 1
|
|
operation = plan.operations[0]
|
|
assert operation.operation_type == "no-op"
|
|
assert operation.destination_path is None
|
|
assert "manual review" in operation.reason.lower()
|
|
|
|
|
|
def test_generate_plan_for_anime_category(config):
|
|
"""Test plan generation for anime files (no-op in v1)."""
|
|
video_file = VideoFile(
|
|
path=Path("/mnt/nas/videos/anime/Some.Anime.01.mkv"),
|
|
filename="Some.Anime.01.mkv",
|
|
size_bytes=500000,
|
|
modified_timestamp=datetime.now(timezone.utc),
|
|
category="anime"
|
|
)
|
|
|
|
# Anime files don't get parsed in v1, so identity is None
|
|
plan = generate_plan([(video_file, None)], config)
|
|
|
|
assert len(plan.operations) == 1
|
|
operation = plan.operations[0]
|
|
assert operation.operation_type == "no-op"
|
|
assert operation.destination_path is None
|
|
assert "anime" in operation.reason.lower() or "not organized" in operation.reason.lower()
|
|
|
|
|
|
def test_generate_plan_for_other_category(config):
|
|
"""Test plan generation for other category files (no-op in v1)."""
|
|
video_file = VideoFile(
|
|
path=Path("/mnt/nas/videos/other/random.mkv"),
|
|
filename="random.mkv",
|
|
size_bytes=500000,
|
|
modified_timestamp=datetime.now(timezone.utc),
|
|
category="other"
|
|
)
|
|
|
|
plan = generate_plan([(video_file, None)], config)
|
|
|
|
assert len(plan.operations) == 1
|
|
operation = plan.operations[0]
|
|
assert operation.operation_type == "no-op"
|
|
assert operation.destination_path is None
|
|
assert "other" in operation.reason.lower() or "not organized" in operation.reason.lower()
|
|
|
|
|
|
def test_generate_plan_preserves_category_boundaries(config):
|
|
"""Test that plan generation preserves category boundaries."""
|
|
movie_file = VideoFile(
|
|
path=Path("/mnt/nas/videos/movie/Movie.2020.mkv"),
|
|
filename="Movie.2020.mkv",
|
|
size_bytes=1000000,
|
|
modified_timestamp=datetime.now(timezone.utc),
|
|
category="movie"
|
|
)
|
|
|
|
movie_identity = MovieIdentity(
|
|
title="Movie",
|
|
year=2020,
|
|
confidence=0.9,
|
|
needs_review=False,
|
|
original_filename="Movie.2020.mkv"
|
|
)
|
|
|
|
plan = generate_plan([(movie_file, movie_identity)], config)
|
|
|
|
operation = plan.operations[0]
|
|
# Destination should still be in movie category
|
|
assert str(operation.destination_path).startswith(str(config.library_root / "movie"))
|
|
|
|
|
|
def test_generate_plan_for_multi_episode_file(config):
|
|
"""Test plan generation for multi-episode files."""
|
|
video_file = VideoFile(
|
|
path=Path("/mnt/nas/videos/series/Show.S01E01E02.mkv"),
|
|
filename="Show.S01E01E02.mkv",
|
|
size_bytes=1000000,
|
|
modified_timestamp=datetime.now(timezone.utc),
|
|
category="series"
|
|
)
|
|
|
|
identity = SeriesIdentity(
|
|
title="Show",
|
|
season=1,
|
|
episodes=[1, 2], # Multi-episode
|
|
confidence=0.9,
|
|
needs_review=False,
|
|
original_filename="Show.S01E01E02.mkv"
|
|
)
|
|
|
|
plan = generate_plan([(video_file, identity)], config)
|
|
|
|
operation = plan.operations[0]
|
|
# Should use first episode number for filename
|
|
assert operation.destination_path == Path("/mnt/nas/videos/series/Show/Season 01/S01E01.mkv")
|
|
|
|
|
|
def test_generate_plan_file_already_at_target(config):
|
|
"""Test plan generation when file is already at target location."""
|
|
# File already in correct location
|
|
video_file = VideoFile(
|
|
path=Path("/mnt/nas/videos/movie/Some Movie (2020)/Some Movie (2020).mkv"),
|
|
filename="Some Movie (2020).mkv",
|
|
size_bytes=1000000,
|
|
modified_timestamp=datetime.now(timezone.utc),
|
|
category="movie"
|
|
)
|
|
|
|
identity = MovieIdentity(
|
|
title="Some Movie",
|
|
year=2020,
|
|
confidence=0.9,
|
|
needs_review=False,
|
|
original_filename="Some Movie (2020).mkv"
|
|
)
|
|
|
|
plan = generate_plan([(video_file, identity)], config)
|
|
|
|
operation = plan.operations[0]
|
|
assert operation.operation_type == "no-op"
|
|
assert "already at target" in operation.reason.lower()
|
|
|
|
|
|
def test_generate_plan_rename_vs_move(config):
|
|
"""Test that plan distinguishes between rename and move operations."""
|
|
# File in correct directory but wrong name (rename)
|
|
video_file_rename = VideoFile(
|
|
path=Path("/mnt/nas/videos/movie/Some Movie (2020)/old_name.mkv"),
|
|
filename="old_name.mkv",
|
|
size_bytes=1000000,
|
|
modified_timestamp=datetime.now(timezone.utc),
|
|
category="movie"
|
|
)
|
|
|
|
identity = MovieIdentity(
|
|
title="Some Movie",
|
|
year=2020,
|
|
confidence=0.9,
|
|
needs_review=False,
|
|
original_filename="old_name.mkv"
|
|
)
|
|
|
|
plan = generate_plan([(video_file_rename, identity)], config)
|
|
operation = plan.operations[0]
|
|
assert operation.operation_type == "rename"
|
|
|
|
# File in wrong directory (move)
|
|
video_file_move = VideoFile(
|
|
path=Path("/mnt/nas/videos/movie/wrong_dir/Some Movie (2020).mkv"),
|
|
filename="Some Movie (2020).mkv",
|
|
size_bytes=1000000,
|
|
modified_timestamp=datetime.now(timezone.utc),
|
|
category="movie"
|
|
)
|
|
|
|
plan = generate_plan([(video_file_move, identity)], config)
|
|
operation = plan.operations[0]
|
|
assert operation.operation_type == "move"
|
|
|
|
|
|
def test_generate_plan_summary(config):
|
|
"""Test that plan summary contains accurate counts."""
|
|
files_and_identities = [
|
|
# Movie with year (move)
|
|
(
|
|
VideoFile(
|
|
path=Path("/mnt/nas/videos/movie/Movie1.2020.mkv"),
|
|
filename="Movie1.2020.mkv",
|
|
size_bytes=1000000,
|
|
modified_timestamp=datetime.now(timezone.utc),
|
|
category="movie"
|
|
),
|
|
MovieIdentity(
|
|
title="Movie1",
|
|
year=2020,
|
|
confidence=0.9,
|
|
needs_review=False,
|
|
original_filename="Movie1.2020.mkv"
|
|
)
|
|
),
|
|
# Movie without year (no-op)
|
|
(
|
|
VideoFile(
|
|
path=Path("/mnt/nas/videos/movie/Movie2.mkv"),
|
|
filename="Movie2.mkv",
|
|
size_bytes=1000000,
|
|
modified_timestamp=datetime.now(timezone.utc),
|
|
category="movie"
|
|
),
|
|
MovieIdentity(
|
|
title="Movie2",
|
|
year=None,
|
|
confidence=0.3,
|
|
needs_review=True,
|
|
original_filename="Movie2.mkv"
|
|
)
|
|
),
|
|
# Anime (no-op)
|
|
(
|
|
VideoFile(
|
|
path=Path("/mnt/nas/videos/anime/Anime.01.mkv"),
|
|
filename="Anime.01.mkv",
|
|
size_bytes=500000,
|
|
modified_timestamp=datetime.now(timezone.utc),
|
|
category="anime"
|
|
),
|
|
None
|
|
),
|
|
# Series with season/episode (move)
|
|
(
|
|
VideoFile(
|
|
path=Path("/mnt/nas/videos/series/Show.S01E01.mkv"),
|
|
filename="Show.S01E01.mkv",
|
|
size_bytes=500000,
|
|
modified_timestamp=datetime.now(timezone.utc),
|
|
category="series"
|
|
),
|
|
SeriesIdentity(
|
|
title="Show",
|
|
season=1,
|
|
episodes=[1],
|
|
confidence=0.9,
|
|
needs_review=False,
|
|
original_filename="Show.S01E01.mkv"
|
|
)
|
|
),
|
|
]
|
|
|
|
plan = generate_plan(files_and_identities, config)
|
|
|
|
assert plan.summary["total"] == 4
|
|
assert plan.summary["move"] == 2
|
|
assert plan.summary["no-op"] == 2
|
|
assert plan.summary["rename"] == 0
|
|
assert plan.summary["quarantine"] == 0
|
|
|
|
|
|
def test_generate_plan_with_analysis_by_quality(config):
|
|
"""With analysis duplicates and by_quality, higher-quality file is kept."""
|
|
config.duplicate_keep = "by_quality"
|
|
p_720 = Path("/mnt/nas/videos/movie/Test.2020.720p.WEB-DL.mkv")
|
|
p_1080 = Path("/mnt/nas/videos/movie/Test.2020.1080p.BluRay.mkv")
|
|
vf_720 = VideoFile(
|
|
path=p_720,
|
|
filename="Test.2020.720p.WEB-DL.mkv",
|
|
size_bytes=1000000,
|
|
modified_timestamp=datetime.now(timezone.utc),
|
|
category="movie",
|
|
)
|
|
vf_1080 = VideoFile(
|
|
path=p_1080,
|
|
filename="Test.2020.1080p.BluRay.mkv",
|
|
size_bytes=2000000,
|
|
modified_timestamp=datetime.now(timezone.utc),
|
|
category="movie",
|
|
)
|
|
identity = MovieIdentity(
|
|
title="Test",
|
|
year=2020,
|
|
confidence=0.9,
|
|
needs_review=False,
|
|
original_filename="Test.2020.mkv",
|
|
)
|
|
identities = [(vf_720, identity), (vf_1080, identity)]
|
|
analysis_data = {
|
|
"metadata": {"source_identities": "identities.json"},
|
|
"completeness": [],
|
|
"duplicates": [
|
|
{
|
|
"identity": {"type": "movie", "title": "Test", "year": 2020},
|
|
"files": [str(p_720), str(p_1080)],
|
|
"quality_comparison": [
|
|
{"path": str(p_720), "resolution": "1280x720", "size_bytes": 1000000},
|
|
{"path": str(p_1080), "resolution": "1920x1080", "size_bytes": 2000000},
|
|
],
|
|
}
|
|
],
|
|
}
|
|
plan = generate_plan(identities, config, analysis_data=analysis_data)
|
|
assert plan.summary["quarantine"] == 1
|
|
assert plan.operations[0].operation_type == "quarantine" # 720p quarantined
|
|
assert plan.operations[1].operation_type == "move" # 1080p kept
|
|
assert "画质" in plan.operations[0].reason
|
|
|
|
|
|
def test_generate_plan_with_analysis_by_reputation_quality_time_reason(config):
|
|
"""New strategy should quarantine duplicates with explicit reason text."""
|
|
config.duplicate_keep = "by_reputation_quality_time"
|
|
p_old = Path("/mnt/nas/videos/movie/Test.2020.720p.WEB-DL.mkv")
|
|
p_new = Path("/mnt/nas/videos/movie/Test.2020.1080p.BluRay.mkv")
|
|
vf_old = VideoFile(
|
|
path=p_old,
|
|
filename="Test.2020.720p.WEB-DL.mkv",
|
|
size_bytes=1000000,
|
|
modified_timestamp=datetime.now(timezone.utc),
|
|
category="movie",
|
|
)
|
|
vf_new = VideoFile(
|
|
path=p_new,
|
|
filename="Test.2020.1080p.BluRay.mkv",
|
|
size_bytes=2000000,
|
|
modified_timestamp=datetime.now(timezone.utc),
|
|
category="movie",
|
|
)
|
|
identity = MovieIdentity(
|
|
title="Test",
|
|
year=2020,
|
|
confidence=0.9,
|
|
needs_review=False,
|
|
original_filename="Test.2020.mkv",
|
|
)
|
|
identities = [(vf_old, identity), (vf_new, identity)]
|
|
analysis_data = {
|
|
"metadata": {"source_identities": "identities.json"},
|
|
"completeness": [],
|
|
"duplicates": [
|
|
{
|
|
"identity": {"type": "movie", "title": "Test", "year": 2020},
|
|
"files": [str(p_old), str(p_new)],
|
|
"quality_comparison": [
|
|
{"path": str(p_old), "resolution": "1280x720", "size_bytes": 1000000},
|
|
{"path": str(p_new), "resolution": "1920x1080", "size_bytes": 2000000},
|
|
],
|
|
}
|
|
],
|
|
}
|
|
plan = generate_plan(identities, config, analysis_data=analysis_data)
|
|
assert plan.summary["quarantine"] == 1
|
|
assert "外部评分>画质>时间" in plan.operations[0].reason
|
|
|
|
|
|
def test_generate_plan_with_analysis_by_reputation_missing_scores_uses_fallback_reason(config):
|
|
"""by_reputation should clearly state quality fallback when reputation is missing."""
|
|
config.duplicate_keep = "by_reputation"
|
|
p_web = Path("/mnt/nas/videos/movie/Test.2020.2160p.WEB-DL.mkv")
|
|
p_bluray = Path("/mnt/nas/videos/movie/Test.2020.1080p.BluRay.mkv")
|
|
vf_web = VideoFile(
|
|
path=p_web,
|
|
filename="Test.2020.2160p.WEB-DL.mkv",
|
|
size_bytes=3000000,
|
|
modified_timestamp=datetime.now(timezone.utc),
|
|
category="movie",
|
|
)
|
|
vf_bluray = VideoFile(
|
|
path=p_bluray,
|
|
filename="Test.2020.1080p.BluRay.mkv",
|
|
size_bytes=2000000,
|
|
modified_timestamp=datetime.now(timezone.utc),
|
|
category="movie",
|
|
)
|
|
identity = MovieIdentity(
|
|
title="Test",
|
|
year=2020,
|
|
confidence=0.9,
|
|
needs_review=False,
|
|
original_filename="Test.2020.mkv",
|
|
)
|
|
identities = [(vf_web, identity), (vf_bluray, identity)]
|
|
analysis_data = {
|
|
"metadata": {"source_identities": "identities.json"},
|
|
"completeness": [],
|
|
"duplicates": [
|
|
{
|
|
"identity": {"type": "movie", "title": "Test", "year": 2020},
|
|
"files": [str(p_web), str(p_bluray)],
|
|
"quality_comparison": [
|
|
{"path": str(p_web), "resolution": "3840x2160", "size_bytes": 3000000},
|
|
{"path": str(p_bluray), "resolution": "1920x1080", "size_bytes": 2000000},
|
|
],
|
|
}
|
|
],
|
|
}
|
|
plan = generate_plan(identities, config, analysis_data=analysis_data)
|
|
assert plan.summary["quarantine"] == 1
|
|
assert "评分缺失/并列" in plan.operations[0].reason
|
|
assert "删除建议（仅隔离建议" in plan.human_summary
|
|
assert "评分依据不足" in plan.human_summary
|
|
|
|
|
|
def test_generate_plan_human_summary_marks_disc_files_as_high_risk(config):
|
|
"""Disc/part files should be listed with a conservative risk note in summary."""
|
|
config.duplicate_keep = "by_reputation"
|
|
p_disc1 = Path("/mnt/nas/videos/movie/The.Best.of.Youth.DISC1.mkv")
|
|
p_disc2 = Path("/mnt/nas/videos/movie/The.Best.of.Youth.DISC2.mkv")
|
|
vf1 = VideoFile(
|
|
path=p_disc1,
|
|
filename="The.Best.of.Youth.DISC1.mkv",
|
|
size_bytes=2000000,
|
|
modified_timestamp=datetime.now(timezone.utc),
|
|
category="movie",
|
|
)
|
|
vf2 = VideoFile(
|
|
path=p_disc2,
|
|
filename="The.Best.of.Youth.DISC2.mkv",
|
|
size_bytes=1800000,
|
|
modified_timestamp=datetime.now(timezone.utc),
|
|
category="movie",
|
|
)
|
|
identity = MovieIdentity(
|
|
title="The Best of Youth",
|
|
year=2003,
|
|
confidence=0.9,
|
|
needs_review=False,
|
|
original_filename="The.Best.of.Youth.DISC1.mkv",
|
|
)
|
|
plan = generate_plan(
|
|
[(vf1, identity), (vf2, identity)],
|
|
config,
|
|
analysis_data={
|
|
"metadata": {"source_identities": "identities.json"},
|
|
"completeness": [],
|
|
"duplicates": [
|
|
{
|
|
"identity": {"type": "movie", "title": "The Best of Youth", "year": 2003},
|
|
"files": [str(p_disc1), str(p_disc2)],
|
|
"quality_comparison": [
|
|
{"path": str(p_disc1), "resolution": "1920x1080", "size_bytes": 2000000},
|
|
{"path": str(p_disc2), "resolution": "1920x1080", "size_bytes": 1800000},
|
|
],
|
|
}
|
|
],
|
|
},
|
|
)
|
|
assert "疑似多碟/分段文件" in plan.human_summary
|
|
|
|
|
|
def test_generate_plan_with_different_extensions(config):
|
|
"""Test plan generation preserves file extensions."""
|
|
extensions = [".mp4", ".mkv", ".avi"]
|
|
|
|
for ext in extensions:
|
|
video_file = VideoFile(
|
|
path=Path(f"/mnt/nas/videos/movie/Movie.2020{ext}"),
|
|
filename=f"Movie.2020{ext}",
|
|
size_bytes=1000000,
|
|
modified_timestamp=datetime.now(timezone.utc),
|
|
category="movie"
|
|
)
|
|
|
|
identity = MovieIdentity(
|
|
title="Movie",
|
|
year=2020,
|
|
confidence=0.9,
|
|
needs_review=False,
|
|
original_filename=f"Movie.2020{ext}"
|
|
)
|
|
|
|
plan = generate_plan([(video_file, identity)], config)
|
|
operation = plan.operations[0]
|
|
|
|
# Check that extension is preserved
|
|
assert operation.destination_path.suffix == ext
|
|
|
|
|
|
def test_conflict_detection_for_movie(config, tmp_path):
|
|
"""Test conflict detection when destination movie file already exists."""
|
|
# Set up config with tmp_path as library root
|
|
config.library_root = tmp_path
|
|
|
|
# Create destination directory and file
|
|
dest_dir = tmp_path / "movie" / "Some Movie (2020)"
|
|
dest_dir.mkdir(parents=True)
|
|
dest_file = dest_dir / "Some Movie (2020).mkv"
|
|
dest_file.touch() # Create the file
|
|
|
|
# Source file in different location
|
|
video_file = VideoFile(
|
|
path=tmp_path / "movie" / "Some.Movie.2020.1080p.mkv",
|
|
filename="Some.Movie.2020.1080p.mkv",
|
|
size_bytes=1000000,
|
|
modified_timestamp=datetime.now(timezone.utc),
|
|
category="movie"
|
|
)
|
|
|
|
identity = MovieIdentity(
|
|
title="Some Movie",
|
|
year=2020,
|
|
confidence=0.9,
|
|
needs_review=False,
|
|
original_filename="Some.Movie.2020.1080p.mkv"
|
|
)
|
|
|
|
plan = generate_plan([(video_file, identity)], config)
|
|
|
|
operation = plan.operations[0]
|
|
assert operation.has_conflict is True
|
|
assert operation.conflict_reason is not None
|
|
assert "already exists" in operation.conflict_reason.lower()
|
|
assert str(dest_file) in operation.conflict_reason
|
|
|
|
|
|
def test_conflict_detection_for_series(config, tmp_path):
|
|
"""Test conflict detection when destination series file already exists."""
|
|
# Set up config with tmp_path as library root
|
|
config.library_root = tmp_path
|
|
|
|
# Create destination directory and file
|
|
dest_dir = tmp_path / "series" / "Show Name" / "Season 01"
|
|
dest_dir.mkdir(parents=True)
|
|
dest_file = dest_dir / "S01E05.mkv"
|
|
dest_file.touch() # Create the file
|
|
|
|
# Source file in different location
|
|
video_file = VideoFile(
|
|
path=tmp_path / "series" / "Show.Name.S01E05.1080p.mkv",
|
|
filename="Show.Name.S01E05.1080p.mkv",
|
|
size_bytes=500000,
|
|
modified_timestamp=datetime.now(timezone.utc),
|
|
category="series"
|
|
)
|
|
|
|
identity = SeriesIdentity(
|
|
title="Show Name",
|
|
season=1,
|
|
episodes=[5],
|
|
confidence=0.9,
|
|
needs_review=False,
|
|
original_filename="Show.Name.S01E05.1080p.mkv"
|
|
)
|
|
|
|
plan = generate_plan([(video_file, identity)], config)
|
|
|
|
operation = plan.operations[0]
|
|
assert operation.has_conflict is True
|
|
assert operation.conflict_reason is not None
|
|
assert "already exists" in operation.conflict_reason.lower()
|
|
assert str(dest_file) in operation.conflict_reason
|
|
|
|
|
|
def test_no_conflict_when_destination_does_not_exist(config, tmp_path):
|
|
"""Test that no conflict is detected when destination file does not exist."""
|
|
# Set up config with tmp_path as library root
|
|
config.library_root = tmp_path
|
|
|
|
# Create source directory but NOT destination
|
|
source_dir = tmp_path / "movie"
|
|
source_dir.mkdir(parents=True)
|
|
|
|
video_file = VideoFile(
|
|
path=source_dir / "Some.Movie.2020.mkv",
|
|
filename="Some.Movie.2020.mkv",
|
|
size_bytes=1000000,
|
|
modified_timestamp=datetime.now(timezone.utc),
|
|
category="movie"
|
|
)
|
|
|
|
identity = MovieIdentity(
|
|
title="Some Movie",
|
|
year=2020,
|
|
confidence=0.9,
|
|
needs_review=False,
|
|
original_filename="Some.Movie.2020.mkv"
|
|
)
|
|
|
|
plan = generate_plan([(video_file, identity)], config)
|
|
|
|
operation = plan.operations[0]
|
|
assert operation.has_conflict is False
|
|
assert operation.conflict_reason is None
|
|
|
|
|
|
def test_conflict_detection_with_multiple_files(config, tmp_path):
|
|
"""Test conflict detection with multiple files, some with conflicts."""
|
|
# Set up config with tmp_path as library root
|
|
config.library_root = tmp_path
|
|
|
|
# Create destination for first movie (conflict)
|
|
dest_dir1 = tmp_path / "movie" / "Movie1 (2020)"
|
|
dest_dir1.mkdir(parents=True)
|
|
(dest_dir1 / "Movie1 (2020).mkv").touch()
|
|
|
|
# Don't create destination for second movie (no conflict)
|
|
|
|
files_and_identities = [
|
|
# Movie 1 - has conflict
|
|
(
|
|
VideoFile(
|
|
path=tmp_path / "movie" / "Movie1.2020.mkv",
|
|
filename="Movie1.2020.mkv",
|
|
size_bytes=1000000,
|
|
modified_timestamp=datetime.now(timezone.utc),
|
|
category="movie"
|
|
),
|
|
MovieIdentity(
|
|
title="Movie1",
|
|
year=2020,
|
|
confidence=0.9,
|
|
needs_review=False,
|
|
original_filename="Movie1.2020.mkv"
|
|
)
|
|
),
|
|
# Movie 2 - no conflict
|
|
(
|
|
VideoFile(
|
|
path=tmp_path / "movie" / "Movie2.2021.mkv",
|
|
filename="Movie2.2021.mkv",
|
|
size_bytes=1000000,
|
|
modified_timestamp=datetime.now(timezone.utc),
|
|
category="movie"
|
|
),
|
|
MovieIdentity(
|
|
title="Movie2",
|
|
year=2021,
|
|
confidence=0.9,
|
|
needs_review=False,
|
|
original_filename="Movie2.2021.mkv"
|
|
)
|
|
),
|
|
]
|
|
|
|
plan = generate_plan(files_and_identities, config)
|
|
|
|
# First operation should have conflict
|
|
assert plan.operations[0].has_conflict is True
|
|
assert plan.operations[0].conflict_reason is not None
|
|
|
|
# Second operation should not have conflict
|
|
assert plan.operations[1].has_conflict is False
|
|
assert plan.operations[1].conflict_reason is None
|
|
|
|
|
|
def test_save_plan_to_json(config, tmp_path):
|
|
"""Test saving execution plan to JSON file."""
|
|
from vlm.planner import save_plan
|
|
|
|
# Create a simple plan
|
|
video_file = VideoFile(
|
|
path=Path("/mnt/nas/videos/movie/Movie.2020.mkv"),
|
|
filename="Movie.2020.mkv",
|
|
size_bytes=1000000,
|
|
modified_timestamp=datetime.now(timezone.utc),
|
|
category="movie"
|
|
)
|
|
|
|
identity = MovieIdentity(
|
|
title="Movie",
|
|
year=2020,
|
|
confidence=0.9,
|
|
needs_review=False,
|
|
original_filename="Movie.2020.mkv"
|
|
)
|
|
|
|
plan = generate_plan([(video_file, identity)], config)
|
|
|
|
# Save to JSON
|
|
output_path = tmp_path / "plan.json"
|
|
save_plan(plan, output_path)
|
|
|
|
# Verify file was created
|
|
assert output_path.exists()
|
|
|
|
# Verify JSON is valid and contains expected fields
|
|
import json
|
|
with open(output_path, 'r') as f:
|
|
plan_dict = json.load(f)
|
|
|
|
assert "plan_id" in plan_dict
|
|
assert "created_at" in plan_dict
|
|
assert "operations" in plan_dict
|
|
assert "summary" in plan_dict
|
|
|
|
assert plan_dict["plan_id"] == plan.plan_id
|
|
assert len(plan_dict["operations"]) == 1
|
|
assert plan_dict["summary"]["total"] == 1
|
|
|
|
|
|
def test_load_plan_from_json(config, tmp_path):
|
|
"""Test loading execution plan from JSON file."""
|
|
from vlm.planner import save_plan, load_plan
|
|
|
|
# Create and save a plan
|
|
video_file = VideoFile(
|
|
path=Path("/mnt/nas/videos/movie/Movie.2020.mkv"),
|
|
filename="Movie.2020.mkv",
|
|
size_bytes=1000000,
|
|
modified_timestamp=datetime.now(timezone.utc),
|
|
category="movie"
|
|
)
|
|
|
|
identity = MovieIdentity(
|
|
title="Movie",
|
|
year=2020,
|
|
confidence=0.9,
|
|
needs_review=False,
|
|
original_filename="Movie.2020.mkv"
|
|
)
|
|
|
|
original_plan = generate_plan([(video_file, identity)], config)
|
|
|
|
output_path = tmp_path / "plan.json"
|
|
save_plan(original_plan, output_path)
|
|
|
|
# Load the plan
|
|
loaded_plan = load_plan(output_path)
|
|
|
|
# Verify loaded plan matches original
|
|
assert loaded_plan.plan_id == original_plan.plan_id
|
|
assert loaded_plan.created_at == original_plan.created_at
|
|
assert len(loaded_plan.operations) == len(original_plan.operations)
|
|
assert loaded_plan.summary == original_plan.summary
|
|
|
|
# Verify operation details
|
|
loaded_op = loaded_plan.operations[0]
|
|
original_op = original_plan.operations[0]
|
|
|
|
assert loaded_op.operation_type == original_op.operation_type
|
|
assert loaded_op.source_path == original_op.source_path
|
|
assert loaded_op.destination_path == original_op.destination_path
|
|
assert loaded_op.reason == original_op.reason
|
|
assert loaded_op.has_conflict == original_op.has_conflict
|
|
assert loaded_op.conflict_reason == original_op.conflict_reason
|
|
|
|
|
|
def test_save_and_load_plan_with_conflicts(config, tmp_path):
|
|
"""Test saving and loading plan with conflict information."""
|
|
from vlm.planner import save_plan, load_plan
|
|
|
|
# Set up config with tmp_path as library root
|
|
config.library_root = tmp_path
|
|
|
|
# Create destination file to trigger conflict
|
|
dest_dir = tmp_path / "movie" / "Movie (2020)"
|
|
dest_dir.mkdir(parents=True)
|
|
(dest_dir / "Movie (2020).mkv").touch()
|
|
|
|
video_file = VideoFile(
|
|
path=tmp_path / "movie" / "Movie.2020.mkv",
|
|
filename="Movie.2020.mkv",
|
|
size_bytes=1000000,
|
|
modified_timestamp=datetime.now(timezone.utc),
|
|
category="movie"
|
|
)
|
|
|
|
identity = MovieIdentity(
|
|
title="Movie",
|
|
year=2020,
|
|
confidence=0.9,
|
|
needs_review=False,
|
|
original_filename="Movie.2020.mkv"
|
|
)
|
|
|
|
original_plan = generate_plan([(video_file, identity)], config)
|
|
|
|
# Verify conflict was detected
|
|
assert original_plan.operations[0].has_conflict is True
|
|
|
|
# Save and load
|
|
output_path = tmp_path / "plan_with_conflict.json"
|
|
save_plan(original_plan, output_path)
|
|
loaded_plan = load_plan(output_path)
|
|
|
|
# Verify conflict information is preserved
|
|
assert loaded_plan.operations[0].has_conflict is True
|
|
assert loaded_plan.operations[0].conflict_reason is not None
|
|
assert "already exists" in loaded_plan.operations[0].conflict_reason.lower()
|
|
|
|
|
|
def test_save_and_load_plan_with_no_op_operations(config, tmp_path):
|
|
"""Test saving and loading plan with no-op operations."""
|
|
from vlm.planner import save_plan, load_plan
|
|
|
|
# Create files that will generate no-op operations
|
|
files_and_identities = [
|
|
# Anime (no-op)
|
|
(
|
|
VideoFile(
|
|
path=Path("/mnt/nas/videos/anime/Anime.01.mkv"),
|
|
filename="Anime.01.mkv",
|
|
size_bytes=500000,
|
|
modified_timestamp=datetime.now(timezone.utc),
|
|
category="anime"
|
|
),
|
|
None
|
|
),
|
|
# Movie without year (no-op)
|
|
(
|
|
VideoFile(
|
|
path=Path("/mnt/nas/videos/movie/Movie.mkv"),
|
|
filename="Movie.mkv",
|
|
size_bytes=1000000,
|
|
modified_timestamp=datetime.now(timezone.utc),
|
|
category="movie"
|
|
),
|
|
MovieIdentity(
|
|
title="Movie",
|
|
year=None,
|
|
confidence=0.3,
|
|
needs_review=True,
|
|
original_filename="Movie.mkv"
|
|
)
|
|
),
|
|
]
|
|
|
|
original_plan = generate_plan(files_and_identities, config)
|
|
|
|
# Save and load
|
|
output_path = tmp_path / "plan_with_noops.json"
|
|
save_plan(original_plan, output_path)
|
|
loaded_plan = load_plan(output_path)
|
|
|
|
# Verify no-op operations are preserved
|
|
assert len(loaded_plan.operations) == 2
|
|
assert all(op.operation_type == "no-op" for op in loaded_plan.operations)
|
|
assert all(op.destination_path is None for op in loaded_plan.operations)
|
|
|
|
|
|
def test_save_plan_json_is_human_readable(config, tmp_path):
|
|
"""Test that saved JSON is human-readable with proper formatting."""
|
|
from vlm.planner import save_plan
|
|
|
|
video_file = VideoFile(
|
|
path=Path("/mnt/nas/videos/movie/Movie.2020.mkv"),
|
|
filename="Movie.2020.mkv",
|
|
size_bytes=1000000,
|
|
modified_timestamp=datetime.now(timezone.utc),
|
|
category="movie"
|
|
)
|
|
|
|
identity = MovieIdentity(
|
|
title="Movie",
|
|
year=2020,
|
|
confidence=0.9,
|
|
needs_review=False,
|
|
original_filename="Movie.2020.mkv"
|
|
)
|
|
|
|
plan = generate_plan([(video_file, identity)], config)
|
|
|
|
output_path = tmp_path / "plan.json"
|
|
save_plan(plan, output_path)
|
|
|
|
# Read the raw JSON content
|
|
with open(output_path, 'r') as f:
|
|
content = f.read()
|
|
|
|
# Verify it's formatted with indentation (human-readable)
|
|
assert "\n" in content # Has newlines
|
|
assert " " in content # Has indentation
|
|
|
|
# Verify it's valid JSON
|
|
import json
|
|
json.loads(content)
|
|
|
|
|
|
def test_save_plan_with_multiple_operations(config, tmp_path):
|
|
"""Test saving plan with multiple operations of different types."""
|
|
from vlm.planner import save_plan, load_plan
|
|
|
|
files_and_identities = [
|
|
# Movie (move)
|
|
(
|
|
VideoFile(
|
|
path=Path("/mnt/nas/videos/movie/Movie1.2020.mkv"),
|
|
filename="Movie1.2020.mkv",
|
|
size_bytes=1000000,
|
|
modified_timestamp=datetime.now(timezone.utc),
|
|
category="movie"
|
|
),
|
|
MovieIdentity(
|
|
title="Movie1",
|
|
year=2020,
|
|
confidence=0.9,
|
|
needs_review=False,
|
|
original_filename="Movie1.2020.mkv"
|
|
)
|
|
),
|
|
# Series (move)
|
|
(
|
|
VideoFile(
|
|
path=Path("/mnt/nas/videos/series/Show.S01E01.mkv"),
|
|
filename="Show.S01E01.mkv",
|
|
size_bytes=500000,
|
|
modified_timestamp=datetime.now(timezone.utc),
|
|
category="series"
|
|
),
|
|
SeriesIdentity(
|
|
title="Show",
|
|
season=1,
|
|
episodes=[1],
|
|
confidence=0.9,
|
|
needs_review=False,
|
|
original_filename="Show.S01E01.mkv"
|
|
)
|
|
),
|
|
# Anime (no-op)
|
|
(
|
|
VideoFile(
|
|
path=Path("/mnt/nas/videos/anime/Anime.01.mkv"),
|
|
filename="Anime.01.mkv",
|
|
size_bytes=500000,
|
|
modified_timestamp=datetime.now(timezone.utc),
|
|
category="anime"
|
|
),
|
|
None
|
|
),
|
|
]
|
|
|
|
original_plan = generate_plan(files_and_identities, config)
|
|
|
|
output_path = tmp_path / "multi_op_plan.json"
|
|
save_plan(original_plan, output_path)
|
|
loaded_plan = load_plan(output_path)
|
|
|
|
# Verify all operations are preserved
|
|
assert len(loaded_plan.operations) == 3
|
|
assert loaded_plan.summary["total"] == 3
|
|
assert loaded_plan.summary["move"] == 2
|
|
assert loaded_plan.summary["no-op"] == 1
|
|
|
|
|
|
def test_load_plan_file_not_found(tmp_path):
|
|
"""Test loading plan from non-existent file raises FileNotFoundError."""
|
|
from vlm.planner import load_plan
|
|
|
|
non_existent_path = tmp_path / "does_not_exist.json"
|
|
|
|
with pytest.raises(FileNotFoundError):
|
|
load_plan(non_existent_path)
|
|
|
|
|
|
def test_load_plan_invalid_json(tmp_path):
|
|
"""Test loading plan from invalid JSON raises JSONDecodeError."""
|
|
from vlm.planner import load_plan
|
|
import json
|
|
|
|
invalid_json_path = tmp_path / "invalid.json"
|
|
with open(invalid_json_path, 'w') as f:
|
|
f.write("{ this is not valid json }")
|
|
|
|
with pytest.raises(json.JSONDecodeError):
|
|
load_plan(invalid_json_path)
|
|
|
|
|
|
def test_plan_json_includes_all_required_fields(config, tmp_path):
|
|
"""Test that saved JSON includes plan_id, created_at, operations, and summary."""
|
|
from vlm.planner import save_plan
|
|
import json
|
|
|
|
video_file = VideoFile(
|
|
path=Path("/mnt/nas/videos/movie/Movie.2020.mkv"),
|
|
filename="Movie.2020.mkv",
|
|
size_bytes=1000000,
|
|
modified_timestamp=datetime.now(timezone.utc),
|
|
category="movie"
|
|
)
|
|
|
|
identity = MovieIdentity(
|
|
title="Movie",
|
|
year=2020,
|
|
confidence=0.9,
|
|
needs_review=False,
|
|
original_filename="Movie.2020.mkv"
|
|
)
|
|
|
|
plan = generate_plan([(video_file, identity)], config)
|
|
|
|
output_path = tmp_path / "plan.json"
|
|
save_plan(plan, output_path)
|
|
|
|
with open(output_path, 'r') as f:
|
|
plan_dict = json.load(f)
|
|
|
|
# Verify all required fields are present
|
|
required_fields = ["plan_id", "created_at", "operations", "summary"]
|
|
for field in required_fields:
|
|
assert field in plan_dict, f"Missing required field: {field}"
|
|
|
|
# Verify operations have required fields
|
|
operation = plan_dict["operations"][0]
|
|
required_op_fields = ["operation_type", "source_path", "destination_path", "reason", "has_conflict", "conflict_reason"]
|
|
for field in required_op_fields:
|
|
assert field in operation, f"Missing required operation field: {field}"
|
|
|
|
|
|
def test_movie_rejected_by_review_generates_noop(config):
|
|
"""Rejected movie should not generate move/rename operation."""
|
|
video_file = VideoFile(
|
|
path=Path("/mnt/nas/videos/movie/Movie.2020.mkv"),
|
|
filename="Movie.2020.mkv",
|
|
size_bytes=1000000,
|
|
modified_timestamp=datetime.now(timezone.utc),
|
|
category="movie"
|
|
)
|
|
|
|
identity = MovieIdentity(
|
|
title="Movie",
|
|
year=2020,
|
|
confidence=0.9,
|
|
needs_review=False,
|
|
original_filename="Movie.2020.mkv",
|
|
review_status="rejected"
|
|
)
|
|
|
|
plan = generate_plan([(video_file, identity)], config)
|
|
assert plan.operations[0].operation_type == "no-op"
|
|
assert "rejected" in plan.operations[0].reason.lower()
|
|
|
|
|
|
def test_series_rejected_by_review_generates_noop(config):
|
|
"""Rejected series should not generate move/rename operation."""
|
|
video_file = VideoFile(
|
|
path=Path("/mnt/nas/videos/series/Show.S01E01.mkv"),
|
|
filename="Show.S01E01.mkv",
|
|
size_bytes=1000000,
|
|
modified_timestamp=datetime.now(timezone.utc),
|
|
category="series"
|
|
)
|
|
|
|
identity = SeriesIdentity(
|
|
title="Show",
|
|
season=1,
|
|
episodes=[1],
|
|
confidence=0.9,
|
|
needs_review=False,
|
|
original_filename="Show.S01E01.mkv",
|
|
review_status="rejected"
|
|
)
|
|
|
|
plan = generate_plan([(video_file, identity)], config)
|
|
assert plan.operations[0].operation_type == "no-op"
|
|
assert "rejected" in plan.operations[0].reason.lower()
|
|
|
|
|
|
def test_save_and_load_plan_validates_schema(tmp_path):
|
|
"""Saved plan artifacts should round-trip through schema validation."""
|
|
plan = ExecutionPlan(
|
|
plan_id="plan-123",
|
|
created_at=datetime(2026, 4, 2, 12, 0, tzinfo=timezone.utc),
|
|
operations=[
|
|
FileOperation(
|
|
operation_type="move",
|
|
source_path=Path("/mnt/nas/videos/movie/source.mkv"),
|
|
destination_path=Path("/mnt/nas/videos/movie/target.mkv"),
|
|
reason="move movie",
|
|
has_conflict=False,
|
|
)
|
|
],
|
|
summary={"move": 1, "rename": 0, "noop": 0, "delete": 0},
|
|
summary_by_reason={"move movie": 1},
|
|
human_summary="计划已生成",
|
|
metadata={"analysis_source": "analysis.json"},
|
|
)
|
|
|
|
output_path = tmp_path / "plan.json"
|
|
from vlm.planner import save_plan
|
|
|
|
save_plan(plan, output_path)
|
|
loaded = load_plan(output_path)
|
|
|
|
assert loaded.plan_id == plan.plan_id
|
|
assert loaded.operations[0].source_path == plan.operations[0].source_path
|
|
assert loaded.operations[0].destination_path == plan.operations[0].destination_path
|
|
assert loaded.summary == plan.summary
|
|
assert loaded.summary_by_reason == plan.summary_by_reason
|
|
assert loaded.human_summary == plan.human_summary
|
|
assert loaded.metadata == plan.metadata
|
|
|
|
|
|
def test_load_plan_rejects_invalid_schema(tmp_path):
|
|
"""Invalid plan artifacts should fail validation before deserialization."""
|
|
invalid_path = tmp_path / "invalid-plan.json"
|
|
invalid_path.write_text(
|
|
json.dumps(
|
|
{
|
|
"vlm_schema_version": "1.0",
|
|
"plan_id": "plan-123",
|
|
"created_at": "2026-04-02T12:00:00+00:00",
|
|
"operations": [
|
|
{
|
|
"operation_type": "move",
|
|
"source_path": "/mnt/nas/videos/movie/source.mkv",
|
|
"destination_path": "/mnt/nas/videos/movie/target.mkv",
|
|
"has_conflict": False,
|
|
}
|
|
],
|
|
"summary": {"move": 1},
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
with pytest.raises(ValueError, match="plan JSON.operations\\[0\\]\\.reason"):
|
|
load_plan(invalid_path)
|
|
|