windyboy
1f55eab304
- Updated `duplicate_resolve.py` to introduce a new strategy for keeping files based on quality, considering resolution, source, codec, and size. - Enhanced `planner.py` to utilize the new quality-based strategy during plan generation, updating quarantine reasons accordingly. - Modified `README.md` to document the new `plan.duplicate_keep` options, including `by_quality`, and provided detailed descriptions of each strategy. - Added unit tests in `test_duplicate_resolve.py` to validate the new quality-based resolution logic. - Updated `analysis.json` and `plan.json` with new timestamps and IDs to reflect recent changes. These updates improve the Video Library Manager's ability to handle duplicate files more effectively, ensuring users retain the highest quality versions.
1021 lines
32 KiB
Python
1021 lines
32 KiB
Python
"""Unit tests for plan generator."""
|
|
|
|
import pytest
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
from vlm.config import Config
|
|
from vlm.models import (
|
|
ExecutionPlan,
|
|
FileOperation,
|
|
MovieIdentity,
|
|
SeriesIdentity,
|
|
VideoFile,
|
|
)
|
|
from vlm.planner import generate_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(),
|
|
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(),
|
|
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(),
|
|
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_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(),
|
|
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(),
|
|
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(),
|
|
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(),
|
|
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(),
|
|
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(),
|
|
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(),
|
|
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(),
|
|
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(),
|
|
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(),
|
|
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(),
|
|
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(),
|
|
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(),
|
|
category="movie",
|
|
)
|
|
vf_1080 = VideoFile(
|
|
path=p_1080,
|
|
filename="Test.2020.1080p.BluRay.mkv",
|
|
size_bytes=2000000,
|
|
modified_timestamp=datetime.now(),
|
|
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_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(),
|
|
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(),
|
|
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(),
|
|
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(),
|
|
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(),
|
|
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(),
|
|
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(),
|
|
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(),
|
|
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(),
|
|
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(),
|
|
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(),
|
|
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(),
|
|
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(),
|
|
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(),
|
|
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(),
|
|
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(),
|
|
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(),
|
|
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(),
|
|
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()
|