910 lines
36 KiB
Python
910 lines
36 KiB
Python
|
"""Tests for quarantine manager."""
|
||
|
|
|
||
|
|
import json
|
||
|
|
import pytest
|
||
|
|
from pathlib import Path
|
||
|
|
from datetime import datetime
|
||
|
|
|
||
|
|
from src.vlm.quarantine import QuarantineManager
|
||
|
|
from src.vlm.config import Config
|
||
|
|
from src.vlm.models import QuarantineEntry, QuarantineManifest
|
||
|
|
|
||
|
|
|
||
|
|
class TestQuarantineManager:
|
||
|
|
"""Test suite for QuarantineManager."""
|
||
|
|
|
||
|
|
@pytest.fixture
|
||
|
|
def config(self, tmp_path):
|
||
|
|
"""Create a test configuration."""
|
||
|
|
library_root = tmp_path / "library"
|
||
|
|
library_root.mkdir()
|
||
|
|
|
||
|
|
# Create category directories
|
||
|
|
(library_root / "movie").mkdir()
|
||
|
|
(library_root / "series").mkdir()
|
||
|
|
(library_root / "anime").mkdir()
|
||
|
|
(library_root / "other").mkdir()
|
||
|
|
|
||
|
|
return Config(
|
||
|
|
library_root=library_root,
|
||
|
|
quarantine_dir=".quarantine"
|
||
|
|
)
|
||
|
|
|
||
|
|
@pytest.fixture
|
||
|
|
def manager(self, config):
|
||
|
|
"""Create a quarantine manager instance."""
|
||
|
|
return QuarantineManager(config)
|
||
|
|
|
||
|
|
def test_quarantine_movie_file(self, manager, config):
|
||
|
|
"""Test quarantining a movie file."""
|
||
|
|
# Create a test movie file
|
||
|
|
movie_file = config.library_root / "movie" / "Test Movie (2020).mkv"
|
||
|
|
movie_file.write_text("test content")
|
||
|
|
|
||
|
|
# Quarantine the file
|
||
|
|
result = manager.quarantine_file(movie_file, reason="duplicate")
|
||
|
|
|
||
|
|
# Verify operation succeeded
|
||
|
|
assert result.success is True
|
||
|
|
assert result.error_message is None
|
||
|
|
|
||
|
|
# Verify file was moved
|
||
|
|
assert not movie_file.exists()
|
||
|
|
|
||
|
|
# Verify file is in quarantine
|
||
|
|
expected_quarantine_path = config.library_root / "movie" / ".quarantine" / "Test Movie (2020).mkv"
|
||
|
|
assert expected_quarantine_path.exists()
|
||
|
|
assert expected_quarantine_path.read_text() == "test content"
|
||
|
|
|
||
|
|
def test_quarantine_series_file(self, manager, config):
|
||
|
|
"""Test quarantining a series file."""
|
||
|
|
# Create a test series file with subdirectory
|
||
|
|
series_dir = config.library_root / "series" / "Test Show" / "Season 01"
|
||
|
|
series_dir.mkdir(parents=True)
|
||
|
|
series_file = series_dir / "S01E01.mkv"
|
||
|
|
series_file.write_text("test content")
|
||
|
|
|
||
|
|
# Quarantine the file
|
||
|
|
result = manager.quarantine_file(series_file, reason="low quality")
|
||
|
|
|
||
|
|
# Verify operation succeeded
|
||
|
|
assert result.success is True
|
||
|
|
assert result.error_message is None
|
||
|
|
|
||
|
|
# Verify file was moved
|
||
|
|
assert not series_file.exists()
|
||
|
|
|
||
|
|
# Verify file is in quarantine with preserved structure
|
||
|
|
expected_quarantine_path = (
|
||
|
|
config.library_root / "series" / ".quarantine" / "Test Show" / "Season 01" / "S01E01.mkv"
|
||
|
|
)
|
||
|
|
assert expected_quarantine_path.exists()
|
||
|
|
assert expected_quarantine_path.read_text() == "test content"
|
||
|
|
|
||
|
|
def test_quarantine_anime_file_rejected(self, manager, config):
|
||
|
|
"""Test that quarantining anime files is rejected."""
|
||
|
|
# Create a test anime file
|
||
|
|
anime_file = config.library_root / "anime" / "Test Anime.mkv"
|
||
|
|
anime_file.write_text("test content")
|
||
|
|
|
||
|
|
# Attempt to quarantine should raise ValueError
|
||
|
|
with pytest.raises(ValueError, match="Quarantine not supported for category 'anime'"):
|
||
|
|
manager.quarantine_file(anime_file)
|
||
|
|
|
||
|
|
# Verify file was not moved
|
||
|
|
assert anime_file.exists()
|
||
|
|
|
||
|
|
def test_quarantine_other_file_rejected(self, manager, config):
|
||
|
|
"""Test that quarantining other files is rejected."""
|
||
|
|
# Create a test other file
|
||
|
|
other_file = config.library_root / "other" / "Test File.mkv"
|
||
|
|
other_file.write_text("test content")
|
||
|
|
|
||
|
|
# Attempt to quarantine should raise ValueError
|
||
|
|
with pytest.raises(ValueError, match="Quarantine not supported for category 'other'"):
|
||
|
|
manager.quarantine_file(other_file)
|
||
|
|
|
||
|
|
# Verify file was not moved
|
||
|
|
assert other_file.exists()
|
||
|
|
|
||
|
|
def test_quarantine_conflict_handling(self, manager, config):
|
||
|
|
"""Test that destination conflicts are handled with numeric suffixes."""
|
||
|
|
# Create a test movie file
|
||
|
|
movie_file = config.library_root / "movie" / "Test Movie (2020).mkv"
|
||
|
|
movie_file.write_text("original content")
|
||
|
|
|
||
|
|
# Create a conflicting file in quarantine
|
||
|
|
quarantine_dir = config.library_root / "movie" / ".quarantine"
|
||
|
|
quarantine_dir.mkdir()
|
||
|
|
existing_file = quarantine_dir / "Test Movie (2020).mkv"
|
||
|
|
existing_file.write_text("existing content")
|
||
|
|
|
||
|
|
# Quarantine the file
|
||
|
|
result = manager.quarantine_file(movie_file)
|
||
|
|
|
||
|
|
# Verify operation succeeded
|
||
|
|
assert result.success is True
|
||
|
|
|
||
|
|
# Verify original file was moved
|
||
|
|
assert not movie_file.exists()
|
||
|
|
|
||
|
|
# Verify existing file is unchanged
|
||
|
|
assert existing_file.exists()
|
||
|
|
assert existing_file.read_text() == "existing content"
|
||
|
|
|
||
|
|
# Verify new file has numeric suffix
|
||
|
|
new_file = quarantine_dir / "Test Movie (2020)_1.mkv"
|
||
|
|
assert new_file.exists()
|
||
|
|
assert new_file.read_text() == "original content"
|
||
|
|
|
||
|
|
def test_quarantine_multiple_conflicts(self, manager, config):
|
||
|
|
"""Test handling multiple conflicts with incrementing suffixes."""
|
||
|
|
# Create quarantine directory with existing files
|
||
|
|
quarantine_dir = config.library_root / "movie" / ".quarantine"
|
||
|
|
quarantine_dir.mkdir()
|
||
|
|
|
||
|
|
# Create existing files
|
||
|
|
(quarantine_dir / "Test Movie (2020).mkv").write_text("file 0")
|
||
|
|
(quarantine_dir / "Test Movie (2020)_1.mkv").write_text("file 1")
|
||
|
|
(quarantine_dir / "Test Movie (2020)_2.mkv").write_text("file 2")
|
||
|
|
|
||
|
|
# Create new file to quarantine
|
||
|
|
movie_file = config.library_root / "movie" / "Test Movie (2020).mkv"
|
||
|
|
movie_file.write_text("file 3")
|
||
|
|
|
||
|
|
# Quarantine the file
|
||
|
|
result = manager.quarantine_file(movie_file)
|
||
|
|
|
||
|
|
# Verify operation succeeded
|
||
|
|
assert result.success is True
|
||
|
|
|
||
|
|
# Verify new file has suffix _3
|
||
|
|
new_file = quarantine_dir / "Test Movie (2020)_3.mkv"
|
||
|
|
assert new_file.exists()
|
||
|
|
assert new_file.read_text() == "file 3"
|
||
|
|
|
||
|
|
def test_quarantine_nonexistent_file(self, manager, config):
|
||
|
|
"""Test quarantining a file that doesn't exist."""
|
||
|
|
# Try to quarantine non-existent file
|
||
|
|
nonexistent_file = config.library_root / "movie" / "Nonexistent.mkv"
|
||
|
|
|
||
|
|
result = manager.quarantine_file(nonexistent_file)
|
||
|
|
|
||
|
|
# Verify operation failed
|
||
|
|
assert result.success is False
|
||
|
|
assert "does not exist" in result.error_message
|
||
|
|
|
||
|
|
def test_quarantine_preserves_directory_structure(self, manager, config):
|
||
|
|
"""Test that quarantine preserves relative directory structure."""
|
||
|
|
# Create a deeply nested series file
|
||
|
|
series_path = config.library_root / "series" / "Show" / "Season 02" / "Extras"
|
||
|
|
series_path.mkdir(parents=True)
|
||
|
|
series_file = series_path / "Behind the Scenes.mkv"
|
||
|
|
series_file.write_text("test content")
|
||
|
|
|
||
|
|
# Quarantine the file
|
||
|
|
result = manager.quarantine_file(series_file)
|
||
|
|
|
||
|
|
# Verify operation succeeded
|
||
|
|
assert result.success is True
|
||
|
|
|
||
|
|
# Verify structure is preserved in quarantine
|
||
|
|
expected_path = (
|
||
|
|
config.library_root / "series" / ".quarantine" /
|
||
|
|
"Show" / "Season 02" / "Extras" / "Behind the Scenes.mkv"
|
||
|
|
)
|
||
|
|
assert expected_path.exists()
|
||
|
|
assert expected_path.read_text() == "test content"
|
||
|
|
|
||
|
|
def test_determine_category_movie(self, manager, config):
|
||
|
|
"""Test category determination for movie files."""
|
||
|
|
movie_file = config.library_root / "movie" / "Test.mkv"
|
||
|
|
category = manager._determine_category(movie_file)
|
||
|
|
assert category == "movie"
|
||
|
|
|
||
|
|
def test_determine_category_series(self, manager, config):
|
||
|
|
"""Test category determination for series files."""
|
||
|
|
series_file = config.library_root / "series" / "Show" / "S01E01.mkv"
|
||
|
|
category = manager._determine_category(series_file)
|
||
|
|
assert category == "series"
|
||
|
|
|
||
|
|
def test_determine_category_anime(self, manager, config):
|
||
|
|
"""Test category determination for anime files."""
|
||
|
|
anime_file = config.library_root / "anime" / "Test.mkv"
|
||
|
|
category = manager._determine_category(anime_file)
|
||
|
|
assert category == "anime"
|
||
|
|
|
||
|
|
def test_determine_category_other(self, manager, config):
|
||
|
|
"""Test category determination for other files."""
|
||
|
|
other_file = config.library_root / "other" / "Test.mkv"
|
||
|
|
category = manager._determine_category(other_file)
|
||
|
|
assert category == "other"
|
||
|
|
|
||
|
|
def test_determine_category_outside_library(self, manager, config):
|
||
|
|
"""Test category determination for files outside library."""
|
||
|
|
outside_file = Path("/tmp/Test.mkv")
|
||
|
|
category = manager._determine_category(outside_file)
|
||
|
|
assert category == "other"
|
||
|
|
|
||
|
|
def test_resolve_conflict_no_conflict(self, manager, config):
|
||
|
|
"""Test conflict resolution when no conflict exists."""
|
||
|
|
test_path = config.library_root / "movie" / ".quarantine" / "Test.mkv"
|
||
|
|
resolved = manager._resolve_conflict(test_path)
|
||
|
|
assert resolved == test_path
|
||
|
|
|
||
|
|
def test_resolve_conflict_with_conflict(self, manager, config):
|
||
|
|
"""Test conflict resolution when conflict exists."""
|
||
|
|
quarantine_dir = config.library_root / "movie" / ".quarantine"
|
||
|
|
quarantine_dir.mkdir(parents=True)
|
||
|
|
|
||
|
|
# Create existing file
|
||
|
|
existing = quarantine_dir / "Test.mkv"
|
||
|
|
existing.write_text("existing")
|
||
|
|
|
||
|
|
# Resolve conflict
|
||
|
|
resolved = manager._resolve_conflict(existing)
|
||
|
|
|
||
|
|
# Should return path with _1 suffix
|
||
|
|
assert resolved == quarantine_dir / "Test_1.mkv"
|
||
|
|
assert not resolved.exists()
|
||
|
|
|
||
|
|
|
||
|
|
class TestQuarantineManifest:
|
||
|
|
"""Test suite for quarantine manifest management."""
|
||
|
|
|
||
|
|
@pytest.fixture
|
||
|
|
def config(self, tmp_path):
|
||
|
|
"""Create a test configuration."""
|
||
|
|
library_root = tmp_path / "library"
|
||
|
|
library_root.mkdir()
|
||
|
|
|
||
|
|
# Create category directories
|
||
|
|
(library_root / "movie").mkdir()
|
||
|
|
(library_root / "series").mkdir()
|
||
|
|
|
||
|
|
return Config(
|
||
|
|
library_root=library_root,
|
||
|
|
quarantine_dir=".quarantine"
|
||
|
|
)
|
||
|
|
|
||
|
|
@pytest.fixture
|
||
|
|
def manager(self, config):
|
||
|
|
"""Create a quarantine manager instance."""
|
||
|
|
return QuarantineManager(config)
|
||
|
|
|
||
|
|
def test_manifest_created_on_first_quarantine(self, manager, config):
|
||
|
|
"""Test that manifest is created when first file is quarantined."""
|
||
|
|
# Create a test movie file
|
||
|
|
movie_file = config.library_root / "movie" / "Test Movie (2020).mkv"
|
||
|
|
movie_file.write_text("test content")
|
||
|
|
|
||
|
|
# Quarantine the file
|
||
|
|
result = manager.quarantine_file(movie_file, reason="duplicate")
|
||
|
|
|
||
|
|
# Verify operation succeeded
|
||
|
|
assert result.success is True
|
||
|
|
|
||
|
|
# Verify manifest was created
|
||
|
|
manifest_path = config.library_root / "movie" / ".quarantine" / "manifest.json"
|
||
|
|
assert manifest_path.exists()
|
||
|
|
|
||
|
|
# Load and verify manifest content
|
||
|
|
with open(manifest_path, 'r') as f:
|
||
|
|
data = json.load(f)
|
||
|
|
|
||
|
|
assert 'entries' in data
|
||
|
|
assert len(data['entries']) == 1
|
||
|
|
|
||
|
|
entry = data['entries'][0]
|
||
|
|
assert entry['original_path'] == str(movie_file)
|
||
|
|
assert entry['category'] == "movie"
|
||
|
|
assert entry['reason'] == "duplicate"
|
||
|
|
assert entry['size_bytes'] == len("test content")
|
||
|
|
assert 'quarantined_at' in entry
|
||
|
|
assert 'quarantine_path' in entry
|
||
|
|
|
||
|
|
def test_manifest_updated_on_subsequent_quarantine(self, manager, config):
|
||
|
|
"""Test that manifest is updated when additional files are quarantined."""
|
||
|
|
# Create and quarantine first file
|
||
|
|
movie_file1 = config.library_root / "movie" / "Movie1.mkv"
|
||
|
|
movie_file1.write_text("content1")
|
||
|
|
manager.quarantine_file(movie_file1, reason="duplicate")
|
||
|
|
|
||
|
|
# Create and quarantine second file
|
||
|
|
movie_file2 = config.library_root / "movie" / "Movie2.mkv"
|
||
|
|
movie_file2.write_text("content2")
|
||
|
|
manager.quarantine_file(movie_file2, reason="low quality")
|
||
|
|
|
||
|
|
# Load manifest
|
||
|
|
manifest_path = config.library_root / "movie" / ".quarantine" / "manifest.json"
|
||
|
|
with open(manifest_path, 'r') as f:
|
||
|
|
data = json.load(f)
|
||
|
|
|
||
|
|
# Verify both entries are in manifest
|
||
|
|
assert len(data['entries']) == 2
|
||
|
|
|
||
|
|
# Verify first entry
|
||
|
|
entry1 = data['entries'][0]
|
||
|
|
assert entry1['original_path'] == str(movie_file1)
|
||
|
|
assert entry1['reason'] == "duplicate"
|
||
|
|
|
||
|
|
# Verify second entry
|
||
|
|
entry2 = data['entries'][1]
|
||
|
|
assert entry2['original_path'] == str(movie_file2)
|
||
|
|
assert entry2['reason'] == "low quality"
|
||
|
|
|
||
|
|
def test_manifest_includes_all_required_fields(self, manager, config):
|
||
|
|
"""Test that manifest entries include all required fields."""
|
||
|
|
# Create a test series file
|
||
|
|
series_dir = config.library_root / "series" / "Show" / "Season 01"
|
||
|
|
series_dir.mkdir(parents=True)
|
||
|
|
series_file = series_dir / "S01E01.mkv"
|
||
|
|
series_file.write_text("test content")
|
||
|
|
|
||
|
|
# Quarantine the file
|
||
|
|
manager.quarantine_file(series_file, reason="test reason")
|
||
|
|
|
||
|
|
# Load manifest
|
||
|
|
manifest_path = config.library_root / "series" / ".quarantine" / "manifest.json"
|
||
|
|
with open(manifest_path, 'r') as f:
|
||
|
|
data = json.load(f)
|
||
|
|
|
||
|
|
entry = data['entries'][0]
|
||
|
|
|
||
|
|
# Verify all required fields are present
|
||
|
|
assert 'original_path' in entry
|
||
|
|
assert 'quarantine_path' in entry
|
||
|
|
assert 'quarantined_at' in entry
|
||
|
|
assert 'reason' in entry
|
||
|
|
assert 'size_bytes' in entry
|
||
|
|
assert 'category' in entry
|
||
|
|
|
||
|
|
# Verify field values
|
||
|
|
assert entry['original_path'] == str(series_file)
|
||
|
|
assert entry['category'] == "series"
|
||
|
|
assert entry['reason'] == "test reason"
|
||
|
|
assert entry['size_bytes'] == len("test content")
|
||
|
|
|
||
|
|
# Verify timestamp is valid ISO format
|
||
|
|
datetime.fromisoformat(entry['quarantined_at'])
|
||
|
|
|
||
|
|
def test_manifest_separate_per_category(self, manager, config):
|
||
|
|
"""Test that each category has its own manifest."""
|
||
|
|
# Create and quarantine movie file
|
||
|
|
movie_file = config.library_root / "movie" / "Movie.mkv"
|
||
|
|
movie_file.write_text("movie content")
|
||
|
|
manager.quarantine_file(movie_file)
|
||
|
|
|
||
|
|
# Create and quarantine series file
|
||
|
|
series_file = config.library_root / "series" / "S01E01.mkv"
|
||
|
|
series_file.write_text("series content")
|
||
|
|
manager.quarantine_file(series_file)
|
||
|
|
|
||
|
|
# Verify separate manifests exist
|
||
|
|
movie_manifest_path = config.library_root / "movie" / ".quarantine" / "manifest.json"
|
||
|
|
series_manifest_path = config.library_root / "series" / ".quarantine" / "manifest.json"
|
||
|
|
|
||
|
|
assert movie_manifest_path.exists()
|
||
|
|
assert series_manifest_path.exists()
|
||
|
|
|
||
|
|
# Verify movie manifest contains only movie entry
|
||
|
|
with open(movie_manifest_path, 'r') as f:
|
||
|
|
movie_data = json.load(f)
|
||
|
|
assert len(movie_data['entries']) == 1
|
||
|
|
assert movie_data['entries'][0]['category'] == "movie"
|
||
|
|
|
||
|
|
# Verify series manifest contains only series entry
|
||
|
|
with open(series_manifest_path, 'r') as f:
|
||
|
|
series_data = json.load(f)
|
||
|
|
assert len(series_data['entries']) == 1
|
||
|
|
assert series_data['entries'][0]['category'] == "series"
|
||
|
|
|
||
|
|
def test_manifest_handles_none_reason(self, manager, config):
|
||
|
|
"""Test that manifest handles files quarantined without a reason."""
|
||
|
|
# Create and quarantine file without reason
|
||
|
|
movie_file = config.library_root / "movie" / "Movie.mkv"
|
||
|
|
movie_file.write_text("content")
|
||
|
|
manager.quarantine_file(movie_file) # No reason provided
|
||
|
|
|
||
|
|
# Load manifest
|
||
|
|
manifest_path = config.library_root / "movie" / ".quarantine" / "manifest.json"
|
||
|
|
with open(manifest_path, 'r') as f:
|
||
|
|
data = json.load(f)
|
||
|
|
|
||
|
|
entry = data['entries'][0]
|
||
|
|
|
||
|
|
# Verify reason is None/null
|
||
|
|
assert entry['reason'] is None
|
||
|
|
|
||
|
|
def test_load_manifest_empty_when_not_exists(self, manager, config):
|
||
|
|
"""Test that loading non-existent manifest returns empty manifest."""
|
||
|
|
manifest = manager._load_manifest("movie")
|
||
|
|
|
||
|
|
assert isinstance(manifest, QuarantineManifest)
|
||
|
|
assert len(manifest.entries) == 0
|
||
|
|
|
||
|
|
def test_load_manifest_parses_existing_manifest(self, manager, config):
|
||
|
|
"""Test that loading existing manifest parses entries correctly."""
|
||
|
|
# Create manifest manually
|
||
|
|
manifest_path = config.library_root / "movie" / ".quarantine" / "manifest.json"
|
||
|
|
manifest_path.parent.mkdir(parents=True)
|
||
|
|
|
||
|
|
test_data = {
|
||
|
|
'entries': [
|
||
|
|
{
|
||
|
|
'original_path': '/path/to/original.mkv',
|
||
|
|
'quarantine_path': '/path/to/quarantine.mkv',
|
||
|
|
'quarantined_at': '2024-01-15T10:30:00',
|
||
|
|
'reason': 'test reason',
|
||
|
|
'size_bytes': 1024,
|
||
|
|
'category': 'movie'
|
||
|
|
}
|
||
|
|
]
|
||
|
|
}
|
||
|
|
|
||
|
|
with open(manifest_path, 'w') as f:
|
||
|
|
json.dump(test_data, f)
|
||
|
|
|
||
|
|
# Load manifest
|
||
|
|
manifest = manager._load_manifest("movie")
|
||
|
|
|
||
|
|
# Verify parsed correctly
|
||
|
|
assert len(manifest.entries) == 1
|
||
|
|
entry = manifest.entries[0]
|
||
|
|
assert entry.original_path == Path('/path/to/original.mkv')
|
||
|
|
assert entry.quarantine_path == Path('/path/to/quarantine.mkv')
|
||
|
|
assert entry.quarantined_at == datetime(2024, 1, 15, 10, 30, 0)
|
||
|
|
assert entry.reason == 'test reason'
|
||
|
|
assert entry.size_bytes == 1024
|
||
|
|
assert entry.category == 'movie'
|
||
|
|
|
||
|
|
def test_save_manifest_creates_directory(self, manager, config):
|
||
|
|
"""Test that saving manifest creates quarantine directory if needed."""
|
||
|
|
# Create manifest
|
||
|
|
manifest = QuarantineManifest(entries=[])
|
||
|
|
|
||
|
|
# Save manifest (directory doesn't exist yet)
|
||
|
|
manager._save_manifest("movie", manifest)
|
||
|
|
|
||
|
|
# Verify directory and file were created
|
||
|
|
manifest_path = config.library_root / "movie" / ".quarantine" / "manifest.json"
|
||
|
|
assert manifest_path.exists()
|
||
|
|
assert manifest_path.parent.is_dir()
|
||
|
|
|
||
|
|
def test_manifest_preserves_utf8_characters(self, manager, config):
|
||
|
|
"""Test that manifest correctly handles UTF-8 characters in paths and reasons."""
|
||
|
|
# Create file with UTF-8 characters
|
||
|
|
movie_file = config.library_root / "movie" / "Café Müller (2020).mkv"
|
||
|
|
movie_file.write_text("content")
|
||
|
|
|
||
|
|
# Quarantine with UTF-8 reason
|
||
|
|
manager.quarantine_file(movie_file, reason="Qualité insuffisante")
|
||
|
|
|
||
|
|
# Load manifest
|
||
|
|
manifest_path = config.library_root / "movie" / ".quarantine" / "manifest.json"
|
||
|
|
with open(manifest_path, 'r', encoding='utf-8') as f:
|
||
|
|
data = json.load(f)
|
||
|
|
|
||
|
|
entry = data['entries'][0]
|
||
|
|
|
||
|
|
# Verify UTF-8 characters are preserved
|
||
|
|
assert "Café Müller" in entry['original_path']
|
||
|
|
assert entry['reason'] == "Qualité insuffisante"
|
||
|
|
|
||
|
|
|
||
|
|
class TestQuarantineListing:
|
||
|
|
"""Test suite for quarantine listing functionality."""
|
||
|
|
|
||
|
|
@pytest.fixture
|
||
|
|
def config(self, tmp_path):
|
||
|
|
"""Create a test configuration."""
|
||
|
|
library_root = tmp_path / "library"
|
||
|
|
library_root.mkdir()
|
||
|
|
|
||
|
|
# Create category directories
|
||
|
|
(library_root / "movie").mkdir()
|
||
|
|
(library_root / "series").mkdir()
|
||
|
|
|
||
|
|
return Config(
|
||
|
|
library_root=library_root,
|
||
|
|
quarantine_dir=".quarantine"
|
||
|
|
)
|
||
|
|
|
||
|
|
@pytest.fixture
|
||
|
|
def manager(self, config):
|
||
|
|
"""Create a quarantine manager instance."""
|
||
|
|
return QuarantineManager(config)
|
||
|
|
|
||
|
|
def test_list_quarantined_empty(self, manager, config):
|
||
|
|
"""Test listing quarantined files when none exist."""
|
||
|
|
entries = manager.list_quarantined()
|
||
|
|
assert entries == []
|
||
|
|
|
||
|
|
def test_list_quarantined_single_category(self, manager, config):
|
||
|
|
"""Test listing quarantined files from a single category."""
|
||
|
|
# Create and quarantine movie files
|
||
|
|
movie1 = config.library_root / "movie" / "Movie1.mkv"
|
||
|
|
movie1.write_text("content1")
|
||
|
|
manager.quarantine_file(movie1, reason="duplicate")
|
||
|
|
|
||
|
|
movie2 = config.library_root / "movie" / "Movie2.mkv"
|
||
|
|
movie2.write_text("content2")
|
||
|
|
manager.quarantine_file(movie2, reason="low quality")
|
||
|
|
|
||
|
|
# List quarantined files from movie category
|
||
|
|
entries = manager.list_quarantined(category="movie")
|
||
|
|
|
||
|
|
assert len(entries) == 2
|
||
|
|
assert all(e.category == "movie" for e in entries)
|
||
|
|
|
||
|
|
# Verify entries contain expected data
|
||
|
|
original_paths = [str(e.original_path) for e in entries]
|
||
|
|
assert str(movie1) in original_paths
|
||
|
|
assert str(movie2) in original_paths
|
||
|
|
|
||
|
|
def test_list_quarantined_all_categories(self, manager, config):
|
||
|
|
"""Test listing quarantined files from all categories."""
|
||
|
|
# Create and quarantine movie file
|
||
|
|
movie = config.library_root / "movie" / "Movie.mkv"
|
||
|
|
movie.write_text("movie content")
|
||
|
|
manager.quarantine_file(movie, reason="duplicate")
|
||
|
|
|
||
|
|
# Create and quarantine series file
|
||
|
|
series = config.library_root / "series" / "S01E01.mkv"
|
||
|
|
series.write_text("series content")
|
||
|
|
manager.quarantine_file(series, reason="low quality")
|
||
|
|
|
||
|
|
# List all quarantined files
|
||
|
|
entries = manager.list_quarantined()
|
||
|
|
|
||
|
|
assert len(entries) == 2
|
||
|
|
|
||
|
|
# Verify both categories are represented
|
||
|
|
categories = [e.category for e in entries]
|
||
|
|
assert "movie" in categories
|
||
|
|
assert "series" in categories
|
||
|
|
|
||
|
|
def test_list_quarantined_filter_by_category(self, manager, config):
|
||
|
|
"""Test filtering quarantined files by category."""
|
||
|
|
# Create and quarantine files in both categories
|
||
|
|
movie = config.library_root / "movie" / "Movie.mkv"
|
||
|
|
movie.write_text("movie content")
|
||
|
|
manager.quarantine_file(movie)
|
||
|
|
|
||
|
|
series = config.library_root / "series" / "S01E01.mkv"
|
||
|
|
series.write_text("series content")
|
||
|
|
manager.quarantine_file(series)
|
||
|
|
|
||
|
|
# List only movie files
|
||
|
|
movie_entries = manager.list_quarantined(category="movie")
|
||
|
|
assert len(movie_entries) == 1
|
||
|
|
assert movie_entries[0].category == "movie"
|
||
|
|
|
||
|
|
# List only series files
|
||
|
|
series_entries = manager.list_quarantined(category="series")
|
||
|
|
assert len(series_entries) == 1
|
||
|
|
assert series_entries[0].category == "series"
|
||
|
|
|
||
|
|
def test_list_quarantined_invalid_category(self, manager, config):
|
||
|
|
"""Test listing with invalid category returns empty list."""
|
||
|
|
entries = manager.list_quarantined(category="anime")
|
||
|
|
assert entries == []
|
||
|
|
|
||
|
|
entries = manager.list_quarantined(category="other")
|
||
|
|
assert entries == []
|
||
|
|
|
||
|
|
entries = manager.list_quarantined(category="invalid")
|
||
|
|
assert entries == []
|
||
|
|
|
||
|
|
def test_list_quarantined_includes_all_fields(self, manager, config):
|
||
|
|
"""Test that listed entries include all required fields."""
|
||
|
|
# Create and quarantine file
|
||
|
|
movie = config.library_root / "movie" / "Movie.mkv"
|
||
|
|
movie.write_text("test content")
|
||
|
|
manager.quarantine_file(movie, reason="test reason")
|
||
|
|
|
||
|
|
# List quarantined files
|
||
|
|
entries = manager.list_quarantined()
|
||
|
|
|
||
|
|
assert len(entries) == 1
|
||
|
|
entry = entries[0]
|
||
|
|
|
||
|
|
# Verify all fields are present
|
||
|
|
assert entry.original_path == movie
|
||
|
|
assert entry.quarantine_path.exists()
|
||
|
|
assert entry.quarantined_at is not None
|
||
|
|
assert entry.reason == "test reason"
|
||
|
|
assert entry.size_bytes == len("test content")
|
||
|
|
assert entry.category == "movie"
|
||
|
|
|
||
|
|
def test_list_quarantined_preserves_directory_structure(self, manager, config):
|
||
|
|
"""Test that listing shows preserved directory structure."""
|
||
|
|
# Create nested series file
|
||
|
|
series_path = config.library_root / "series" / "Show" / "Season 01"
|
||
|
|
series_path.mkdir(parents=True)
|
||
|
|
series_file = series_path / "S01E01.mkv"
|
||
|
|
series_file.write_text("content")
|
||
|
|
|
||
|
|
# Quarantine the file
|
||
|
|
manager.quarantine_file(series_file)
|
||
|
|
|
||
|
|
# List quarantined files
|
||
|
|
entries = manager.list_quarantined(category="series")
|
||
|
|
|
||
|
|
assert len(entries) == 1
|
||
|
|
entry = entries[0]
|
||
|
|
|
||
|
|
# Verify quarantine path preserves structure
|
||
|
|
expected_quarantine = (
|
||
|
|
config.library_root / "series" / ".quarantine" /
|
||
|
|
"Show" / "Season 01" / "S01E01.mkv"
|
||
|
|
)
|
||
|
|
assert entry.quarantine_path == expected_quarantine
|
||
|
|
|
||
|
|
def test_list_quarantined_multiple_files_same_category(self, manager, config):
|
||
|
|
"""Test listing multiple files from the same category."""
|
||
|
|
# Create and quarantine multiple movie files
|
||
|
|
for i in range(5):
|
||
|
|
movie = config.library_root / "movie" / f"Movie{i}.mkv"
|
||
|
|
movie.write_text(f"content{i}")
|
||
|
|
manager.quarantine_file(movie, reason=f"reason{i}")
|
||
|
|
|
||
|
|
# List quarantined files
|
||
|
|
entries = manager.list_quarantined(category="movie")
|
||
|
|
|
||
|
|
assert len(entries) == 5
|
||
|
|
|
||
|
|
# Verify all entries are unique
|
||
|
|
original_paths = [e.original_path for e in entries]
|
||
|
|
assert len(set(original_paths)) == 5
|
||
|
|
|
||
|
|
|
||
|
|
class TestQuarantineRestoration:
|
||
|
|
"""Test suite for quarantine restoration functionality."""
|
||
|
|
|
||
|
|
@pytest.fixture
|
||
|
|
def config(self, tmp_path):
|
||
|
|
"""Create a test configuration."""
|
||
|
|
library_root = tmp_path / "library"
|
||
|
|
library_root.mkdir()
|
||
|
|
|
||
|
|
# Create category directories
|
||
|
|
(library_root / "movie").mkdir()
|
||
|
|
(library_root / "series").mkdir()
|
||
|
|
|
||
|
|
return Config(
|
||
|
|
library_root=library_root,
|
||
|
|
quarantine_dir=".quarantine"
|
||
|
|
)
|
||
|
|
|
||
|
|
@pytest.fixture
|
||
|
|
def manager(self, config):
|
||
|
|
"""Create a quarantine manager instance."""
|
||
|
|
return QuarantineManager(config)
|
||
|
|
|
||
|
|
def test_restore_movie_file(self, manager, config):
|
||
|
|
"""Test restoring a quarantined movie file."""
|
||
|
|
# Create and quarantine movie file
|
||
|
|
original_path = config.library_root / "movie" / "Movie.mkv"
|
||
|
|
original_path.write_text("test content")
|
||
|
|
|
||
|
|
result = manager.quarantine_file(original_path, reason="duplicate")
|
||
|
|
assert result.success is True
|
||
|
|
|
||
|
|
quarantine_path = result.operation.destination_path
|
||
|
|
|
||
|
|
# Verify file is in quarantine
|
||
|
|
assert quarantine_path.exists()
|
||
|
|
assert not original_path.exists()
|
||
|
|
|
||
|
|
# Restore the file
|
||
|
|
restore_result = manager.restore_from_quarantine(quarantine_path)
|
||
|
|
|
||
|
|
# Verify restoration succeeded
|
||
|
|
assert restore_result.success is True
|
||
|
|
assert restore_result.error_message is None
|
||
|
|
|
||
|
|
# Verify file is back at original location
|
||
|
|
assert original_path.exists()
|
||
|
|
assert original_path.read_text() == "test content"
|
||
|
|
|
||
|
|
# Verify file is removed from quarantine
|
||
|
|
assert not quarantine_path.exists()
|
||
|
|
|
||
|
|
def test_restore_series_file(self, manager, config):
|
||
|
|
"""Test restoring a quarantined series file."""
|
||
|
|
# Create nested series file
|
||
|
|
series_path = config.library_root / "series" / "Show" / "Season 01"
|
||
|
|
series_path.mkdir(parents=True)
|
||
|
|
original_path = series_path / "S01E01.mkv"
|
||
|
|
original_path.write_text("series content")
|
||
|
|
|
||
|
|
# Quarantine the file
|
||
|
|
result = manager.quarantine_file(original_path)
|
||
|
|
quarantine_path = result.operation.destination_path
|
||
|
|
|
||
|
|
# Restore the file
|
||
|
|
restore_result = manager.restore_from_quarantine(quarantine_path)
|
||
|
|
|
||
|
|
# Verify restoration succeeded
|
||
|
|
assert restore_result.success is True
|
||
|
|
|
||
|
|
# Verify file is back with preserved structure
|
||
|
|
assert original_path.exists()
|
||
|
|
assert original_path.read_text() == "series content"
|
||
|
|
assert not quarantine_path.exists()
|
||
|
|
|
||
|
|
def test_restore_removes_manifest_entry(self, manager, config):
|
||
|
|
"""Test that restoration removes entry from manifest."""
|
||
|
|
# Create and quarantine file
|
||
|
|
original_path = config.library_root / "movie" / "Movie.mkv"
|
||
|
|
original_path.write_text("content")
|
||
|
|
|
||
|
|
result = manager.quarantine_file(original_path)
|
||
|
|
quarantine_path = result.operation.destination_path
|
||
|
|
|
||
|
|
# Verify manifest has entry
|
||
|
|
manifest = manager._load_manifest("movie")
|
||
|
|
assert len(manifest.entries) == 1
|
||
|
|
|
||
|
|
# Restore the file
|
||
|
|
manager.restore_from_quarantine(quarantine_path)
|
||
|
|
|
||
|
|
# Verify manifest entry is removed
|
||
|
|
manifest = manager._load_manifest("movie")
|
||
|
|
assert len(manifest.entries) == 0
|
||
|
|
|
||
|
|
def test_restore_nonexistent_file(self, manager, config):
|
||
|
|
"""Test restoring a file that doesn't exist."""
|
||
|
|
# Try to restore non-existent file
|
||
|
|
nonexistent = config.library_root / "movie" / ".quarantine" / "Nonexistent.mkv"
|
||
|
|
|
||
|
|
result = manager.restore_from_quarantine(nonexistent)
|
||
|
|
|
||
|
|
# Verify operation failed
|
||
|
|
assert result.success is False
|
||
|
|
assert "does not exist" in result.error_message
|
||
|
|
|
||
|
|
def test_restore_conflict_destination_exists(self, manager, config):
|
||
|
|
"""Test restoration when original location already has a file."""
|
||
|
|
# Create and quarantine file
|
||
|
|
original_path = config.library_root / "movie" / "Movie.mkv"
|
||
|
|
original_path.write_text("original content")
|
||
|
|
|
||
|
|
result = manager.quarantine_file(original_path)
|
||
|
|
quarantine_path = result.operation.destination_path
|
||
|
|
|
||
|
|
# Create a new file at original location
|
||
|
|
original_path.write_text("new content")
|
||
|
|
|
||
|
|
# Try to restore
|
||
|
|
restore_result = manager.restore_from_quarantine(quarantine_path)
|
||
|
|
|
||
|
|
# Verify restoration failed due to conflict
|
||
|
|
assert restore_result.success is False
|
||
|
|
assert "already exists" in restore_result.error_message
|
||
|
|
assert restore_result.operation.has_conflict is True
|
||
|
|
|
||
|
|
# Verify original file is unchanged
|
||
|
|
assert original_path.read_text() == "new content"
|
||
|
|
|
||
|
|
# Verify quarantine file still exists
|
||
|
|
assert quarantine_path.exists()
|
||
|
|
|
||
|
|
def test_restore_creates_parent_directory(self, manager, config):
|
||
|
|
"""Test that restoration creates parent directory if needed."""
|
||
|
|
# Create and quarantine file
|
||
|
|
series_path = config.library_root / "series" / "Show" / "Season 01"
|
||
|
|
series_path.mkdir(parents=True)
|
||
|
|
original_path = series_path / "S01E01.mkv"
|
||
|
|
original_path.write_text("content")
|
||
|
|
|
||
|
|
result = manager.quarantine_file(original_path)
|
||
|
|
quarantine_path = result.operation.destination_path
|
||
|
|
|
||
|
|
# Remove the parent directory
|
||
|
|
import shutil
|
||
|
|
shutil.rmtree(series_path)
|
||
|
|
|
||
|
|
# Restore the file
|
||
|
|
restore_result = manager.restore_from_quarantine(quarantine_path)
|
||
|
|
|
||
|
|
# Verify restoration succeeded and directory was created
|
||
|
|
assert restore_result.success is True
|
||
|
|
assert original_path.exists()
|
||
|
|
assert original_path.parent.is_dir()
|
||
|
|
|
||
|
|
def test_restore_no_manifest_entry(self, manager, config):
|
||
|
|
"""Test restoring a file that has no manifest entry."""
|
||
|
|
# Create quarantine file manually without manifest entry
|
||
|
|
quarantine_dir = config.library_root / "movie" / ".quarantine"
|
||
|
|
quarantine_dir.mkdir(parents=True)
|
||
|
|
quarantine_path = quarantine_dir / "Movie.mkv"
|
||
|
|
quarantine_path.write_text("content")
|
||
|
|
|
||
|
|
# Try to restore
|
||
|
|
result = manager.restore_from_quarantine(quarantine_path)
|
||
|
|
|
||
|
|
# Verify operation failed
|
||
|
|
assert result.success is False
|
||
|
|
assert "No manifest entry" in result.error_message
|
||
|
|
|
||
|
|
def test_restore_invalid_quarantine_path(self, manager, config):
|
||
|
|
"""Test restoring from an invalid quarantine path."""
|
||
|
|
# Try to restore from non-quarantine location
|
||
|
|
invalid_path = config.library_root / "movie" / "Movie.mkv"
|
||
|
|
invalid_path.write_text("content")
|
||
|
|
|
||
|
|
result = manager.restore_from_quarantine(invalid_path)
|
||
|
|
|
||
|
|
# Verify operation failed
|
||
|
|
assert result.success is False
|
||
|
|
assert "Could not determine category" in result.error_message
|
||
|
|
|
||
|
|
def test_restore_multiple_files(self, manager, config):
|
||
|
|
"""Test restoring multiple quarantined files."""
|
||
|
|
# Create and quarantine multiple files
|
||
|
|
files = []
|
||
|
|
quarantine_paths = []
|
||
|
|
|
||
|
|
for i in range(3):
|
||
|
|
file_path = config.library_root / "movie" / f"Movie{i}.mkv"
|
||
|
|
file_path.write_text(f"content{i}")
|
||
|
|
files.append(file_path)
|
||
|
|
|
||
|
|
result = manager.quarantine_file(file_path)
|
||
|
|
quarantine_paths.append(result.operation.destination_path)
|
||
|
|
|
||
|
|
# Verify all files are quarantined
|
||
|
|
for file_path in files:
|
||
|
|
assert not file_path.exists()
|
||
|
|
for qpath in quarantine_paths:
|
||
|
|
assert qpath.exists()
|
||
|
|
|
||
|
|
# Restore all files
|
||
|
|
for qpath in quarantine_paths:
|
||
|
|
result = manager.restore_from_quarantine(qpath)
|
||
|
|
assert result.success is True
|
||
|
|
|
||
|
|
# Verify all files are restored
|
||
|
|
for i, file_path in enumerate(files):
|
||
|
|
assert file_path.exists()
|
||
|
|
assert file_path.read_text() == f"content{i}"
|
||
|
|
|
||
|
|
# Verify all quarantine files are removed
|
||
|
|
for qpath in quarantine_paths:
|
||
|
|
assert not qpath.exists()
|
||
|
|
|
||
|
|
# Verify manifest is empty
|
||
|
|
manifest = manager._load_manifest("movie")
|
||
|
|
assert len(manifest.entries) == 0
|
||
|
|
|
||
|
|
def test_determine_category_from_quarantine_movie(self, manager, config):
|
||
|
|
"""Test determining category from movie quarantine path."""
|
||
|
|
qpath = config.library_root / "movie" / ".quarantine" / "Movie.mkv"
|
||
|
|
category = manager._determine_category_from_quarantine(qpath)
|
||
|
|
assert category == "movie"
|
||
|
|
|
||
|
|
def test_determine_category_from_quarantine_series(self, manager, config):
|
||
|
|
"""Test determining category from series quarantine path."""
|
||
|
|
qpath = config.library_root / "series" / ".quarantine" / "Show" / "S01E01.mkv"
|
||
|
|
category = manager._determine_category_from_quarantine(qpath)
|
||
|
|
assert category == "series"
|
||
|
|
|
||
|
|
def test_determine_category_from_quarantine_invalid(self, manager, config):
|
||
|
|
"""Test determining category from invalid quarantine path."""
|
||
|
|
# Not in quarantine directory
|
||
|
|
invalid = config.library_root / "movie" / "Movie.mkv"
|
||
|
|
category = manager._determine_category_from_quarantine(invalid)
|
||
|
|
assert category is None
|
||
|
|
|
||
|
|
# Outside library root
|
||
|
|
outside = Path("/tmp/Movie.mkv")
|
||
|
|
category = manager._determine_category_from_quarantine(outside)
|
||
|
|
assert category is None
|
||
|
|
|
||
|
|
# Unsupported category
|
||
|
|
anime = config.library_root / "anime" / ".quarantine" / "Anime.mkv"
|
||
|
|
category = manager._determine_category_from_quarantine(anime)
|
||
|
|
assert category is None
|