Lukas
239341629c
- Add _cleanup_orphaned_folder() to delete/move old folder contents after rename - Empty folders: delete directly via rmdir() - Non-empty folders: move contents to new path, then delete old folder - Handle PermissionError and OSError gracefully with logging - Add dry_run parameter to preview changes without applying them - Add --dry-run support to validate_and_rename_series_folders() - Add unit tests for _cleanup_orphaned_folder and dry-run mode - All 66 related tests pass Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
555 lines
21 KiB
Python
555 lines
21 KiB
Python
"""Unit tests for folder_rename_service.py.
|
|
|
|
These tests verify the core logic of the folder rename service in
|
|
isolation, using temporary directories and mocked dependencies.
|
|
"""
|
|
from pathlib import Path
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from src.server.services.folder_rename_service import (
|
|
_cleanup_orphaned_folder,
|
|
_compute_expected_folder_name,
|
|
_is_series_being_downloaded,
|
|
_parse_nfo_title_and_year,
|
|
_update_database_paths,
|
|
validate_and_rename_series_folders,
|
|
)
|
|
|
|
|
|
class TestParseNfoTitleAndYear:
|
|
"""Tests for _parse_nfo_title_and_year."""
|
|
|
|
def test_parses_title_and_year(self, tmp_path: Path) -> None:
|
|
nfo = tmp_path / "tvshow.nfo"
|
|
nfo.write_text(
|
|
"<tvshow><title>Attack on Titan</title><year>2013</year></tvshow>"
|
|
)
|
|
title, year = _parse_nfo_title_and_year(nfo)
|
|
assert title == "Attack on Titan"
|
|
assert year == "2013"
|
|
|
|
def test_missing_title_returns_none(self, tmp_path: Path) -> None:
|
|
nfo = tmp_path / "tvshow.nfo"
|
|
nfo.write_text("<tvshow><year>2013</year></tvshow>")
|
|
title, year = _parse_nfo_title_and_year(nfo)
|
|
assert title is None
|
|
assert year == "2013"
|
|
|
|
def test_missing_year_returns_none(self, tmp_path: Path) -> None:
|
|
nfo = tmp_path / "tvshow.nfo"
|
|
nfo.write_text("<tvshow><title>Attack on Titan</title></tvshow>")
|
|
title, year = _parse_nfo_title_and_year(nfo)
|
|
assert title == "Attack on Titan"
|
|
assert year is None
|
|
|
|
def test_empty_title_returns_none(self, tmp_path: Path) -> None:
|
|
nfo = tmp_path / "tvshow.nfo"
|
|
nfo.write_text(
|
|
"<tvshow><title> </title><year>2013</year></tvshow>"
|
|
)
|
|
title, year = _parse_nfo_title_and_year(nfo)
|
|
assert title is None
|
|
assert year == "2013"
|
|
|
|
def test_malformed_xml_returns_none(self, tmp_path: Path) -> None:
|
|
nfo = tmp_path / "tvshow.nfo"
|
|
nfo.write_text("not xml at all")
|
|
title, year = _parse_nfo_title_and_year(nfo)
|
|
assert title is None
|
|
assert year is None
|
|
|
|
|
|
class TestComputeExpectedFolderName:
|
|
"""Tests for _compute_expected_folder_name."""
|
|
|
|
def test_simple_title_and_year(self) -> None:
|
|
result = _compute_expected_folder_name("Attack on Titan", "2013")
|
|
assert result == "Attack on Titan (2013)"
|
|
|
|
def test_sanitizes_invalid_chars(self) -> None:
|
|
result = _compute_expected_folder_name("Show: Subtitle", "2020")
|
|
assert result == "Show Subtitle (2020)"
|
|
|
|
def test_sanitizes_slashes(self) -> None:
|
|
result = _compute_expected_folder_name("A / B", "2021")
|
|
assert result == "A B (2021)"
|
|
|
|
def test_does_not_duplicate_year(self) -> None:
|
|
result = _compute_expected_folder_name("86 Eighty Six (2021)", "2021")
|
|
assert result == "86 Eighty Six (2021)"
|
|
assert result.count("(2021)") == 1
|
|
|
|
def test_removes_duplicate_year_suffixes_bug_86_eighty_six(self) -> None:
|
|
"""Test the bug fix for duplicate year suffixes.
|
|
|
|
Issue: "86 Eighty Six (2021) (2021) (2021) (2021) (2021)"
|
|
should become "86 Eighty Six (2021)"
|
|
"""
|
|
result = _compute_expected_folder_name(
|
|
"86 Eighty Six (2021) (2021) (2021) (2021) (2021)", "2021"
|
|
)
|
|
assert result == "86 Eighty Six (2021)"
|
|
assert result.count("(2021)") == 1
|
|
|
|
def test_removes_duplicate_year_suffixes_alma_chan(self) -> None:
|
|
"""Test the bug fix for duplicate year suffixes with long title.
|
|
|
|
Issue: "Alma-chan Wants to Be a Family! (2025) (2025) (2025) (2025) (2025)"
|
|
should become "Alma-chan Wants to Be a Family! (2025)"
|
|
"""
|
|
result = _compute_expected_folder_name(
|
|
"Alma-chan Wants to Be a Family! (2025) (2025) (2025) (2025) (2025)",
|
|
"2025",
|
|
)
|
|
assert result == "Alma-chan Wants to Be a Family! (2025)"
|
|
assert result.count("(2025)") == 1
|
|
|
|
def test_removes_duplicate_year_suffixes_bogus_skill(self) -> None:
|
|
"""Test the bug fix for duplicate year suffixes with very long title.
|
|
|
|
Issue: Long title with duplicated years should be cleaned.
|
|
"""
|
|
result = _compute_expected_folder_name(
|
|
"Bogus Skill Fruitmaster About That Time I Became Able to Eat "
|
|
"Unlimited Numbers of Skill Fruits (That Kill You) (2025) (2025)",
|
|
"2025",
|
|
)
|
|
assert "(2025)" in result
|
|
assert result.count("(2025)") == 1
|
|
|
|
def test_removes_multiple_different_year_suffixes(self) -> None:
|
|
"""Test that old duplicate years are removed and new one added."""
|
|
result = _compute_expected_folder_name(
|
|
"Series (2020) (2020) (2020)", "2021"
|
|
)
|
|
assert result == "Series (2021)"
|
|
assert "(2020)" not in result
|
|
assert result.count("(2021)") == 1
|
|
|
|
def test_handles_whitespace_with_duplicate_years(self) -> None:
|
|
"""Test that extra whitespace is removed along with duplicate years."""
|
|
result = _compute_expected_folder_name(
|
|
"Series (2021) (2021) (2021) ", "2021"
|
|
)
|
|
assert result == "Series (2021)"
|
|
assert result.count("(2021)") == 1
|
|
assert not result.endswith(" ")
|
|
|
|
def test_idempotent_multiple_calls(self) -> None:
|
|
"""Test that calling the function multiple times produces the same result."""
|
|
title = "86 Eighty Six (2021) (2021) (2021)"
|
|
year = "2021"
|
|
|
|
# First call
|
|
result1 = _compute_expected_folder_name(title, year)
|
|
# Second call with the result
|
|
result2 = _compute_expected_folder_name(result1, year)
|
|
# Third call with the result
|
|
result3 = _compute_expected_folder_name(result2, year)
|
|
|
|
# All results should be identical
|
|
assert result1 == result2 == result3
|
|
assert result1 == "86 Eighty Six (2021)"
|
|
assert result1.count("(2021)") == 1
|
|
|
|
|
|
class TestIsSeriesBeingDownloaded:
|
|
"""Tests for _is_series_being_downloaded."""
|
|
|
|
def test_no_active_download(self) -> None:
|
|
mock_service = MagicMock()
|
|
mock_service._active_download = None
|
|
mock_service._pending_queue = []
|
|
with patch(
|
|
"src.server.services.folder_rename_service.get_download_service",
|
|
return_value=mock_service,
|
|
):
|
|
assert _is_series_being_downloaded("Some Show") is False
|
|
|
|
def test_active_download_matches(self) -> None:
|
|
mock_item = MagicMock()
|
|
mock_item.serie_folder = "Some Show"
|
|
mock_service = MagicMock()
|
|
mock_service._active_download = mock_item
|
|
mock_service._pending_queue = []
|
|
with patch(
|
|
"src.server.services.folder_rename_service.get_download_service",
|
|
return_value=mock_service,
|
|
):
|
|
assert _is_series_being_downloaded("Some Show") is True
|
|
|
|
def test_pending_download_matches(self) -> None:
|
|
mock_item = MagicMock()
|
|
mock_item.serie_folder = "Some Show"
|
|
mock_service = MagicMock()
|
|
mock_service._active_download = None
|
|
mock_service._pending_queue = [mock_item]
|
|
with patch(
|
|
"src.server.services.folder_rename_service.get_download_service",
|
|
return_value=mock_service,
|
|
):
|
|
assert _is_series_being_downloaded("Some Show") is True
|
|
|
|
def test_exception_returns_true_for_safety(self) -> None:
|
|
with patch(
|
|
"src.server.services.folder_rename_service.get_download_service",
|
|
side_effect=RuntimeError("boom"),
|
|
):
|
|
assert _is_series_being_downloaded("Some Show") is True
|
|
|
|
|
|
class TestUpdateDatabasePaths:
|
|
"""Tests for _update_database_paths."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_updates_series_folder(self, tmp_path: Path) -> None:
|
|
anime_dir = tmp_path / "anime"
|
|
anime_dir.mkdir()
|
|
|
|
mock_series = MagicMock()
|
|
mock_series.id = 1
|
|
mock_series.folder = "Old Name"
|
|
|
|
with patch(
|
|
"src.server.services.folder_rename_service.get_db_session"
|
|
) as mock_get_db, patch(
|
|
"src.server.services.folder_rename_service.AnimeSeriesService"
|
|
) as mock_series_svc, patch(
|
|
"src.server.services.folder_rename_service.EpisodeService"
|
|
) as mock_episode_svc, patch(
|
|
"src.server.services.folder_rename_service.DownloadQueueService"
|
|
) as mock_queue_svc:
|
|
|
|
mock_db = AsyncMock()
|
|
mock_get_db.return_value.__aenter__ = AsyncMock(return_value=mock_db)
|
|
mock_get_db.return_value.__aexit__ = AsyncMock(return_value=False)
|
|
|
|
mock_series_svc.get_by_key = AsyncMock(return_value=mock_series)
|
|
mock_series_svc.get_all = AsyncMock(return_value=[])
|
|
mock_series_svc.update = AsyncMock(return_value=mock_series)
|
|
|
|
mock_episode_svc.get_by_series = AsyncMock(return_value=[])
|
|
mock_queue_svc.get_all = AsyncMock(return_value=[])
|
|
|
|
await _update_database_paths("Old Name", "New Name", anime_dir)
|
|
|
|
mock_series_svc.update.assert_awaited_once_with(
|
|
mock_db, 1, folder="New Name"
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_updates_episode_file_paths(self, tmp_path: Path) -> None:
|
|
anime_dir = tmp_path / "anime"
|
|
anime_dir.mkdir()
|
|
old_path = anime_dir / "Old Name" / "S01E01.mkv"
|
|
new_path = anime_dir / "New Name" / "S01E01.mkv"
|
|
|
|
mock_series = MagicMock()
|
|
mock_series.id = 1
|
|
mock_series.folder = "Old Name"
|
|
|
|
mock_episode = MagicMock()
|
|
mock_episode.file_path = str(old_path)
|
|
|
|
with patch(
|
|
"src.server.services.folder_rename_service.get_db_session"
|
|
) as mock_get_db, patch(
|
|
"src.server.services.folder_rename_service.AnimeSeriesService"
|
|
) as mock_series_svc, patch(
|
|
"src.server.services.folder_rename_service.EpisodeService"
|
|
) as mock_episode_svc, patch(
|
|
"src.server.services.folder_rename_service.DownloadQueueService"
|
|
) as mock_queue_svc:
|
|
|
|
mock_db = AsyncMock()
|
|
mock_get_db.return_value.__aenter__ = AsyncMock(return_value=mock_db)
|
|
mock_get_db.return_value.__aexit__ = AsyncMock(return_value=False)
|
|
|
|
mock_series_svc.get_by_key = AsyncMock(return_value=mock_series)
|
|
mock_series_svc.get_all = AsyncMock(return_value=[])
|
|
mock_series_svc.update = AsyncMock(return_value=mock_series)
|
|
|
|
mock_episode_svc.get_by_series = AsyncMock(return_value=[mock_episode])
|
|
mock_queue_svc.get_all = AsyncMock(return_value=[])
|
|
|
|
await _update_database_paths("Old Name", "New Name", anime_dir)
|
|
|
|
assert mock_episode.file_path == str(new_path)
|
|
|
|
|
|
class TestCleanupOrphanedFolder:
|
|
"""Tests for _cleanup_orphaned_folder."""
|
|
|
|
def test_returns_false_when_old_folder_does_not_exist(self, tmp_path: Path) -> None:
|
|
old_path = tmp_path / "nonexistent"
|
|
new_path = tmp_path / "new"
|
|
result = _cleanup_orphaned_folder(old_path, new_path)
|
|
assert result is False
|
|
|
|
def test_deletes_empty_folder(self, tmp_path: Path) -> None:
|
|
old_path = tmp_path / "empty_orphan"
|
|
old_path.mkdir()
|
|
new_path = tmp_path / "new"
|
|
new_path.mkdir()
|
|
result = _cleanup_orphaned_folder(old_path, new_path)
|
|
assert result is True
|
|
assert not old_path.exists()
|
|
|
|
def test_moves_files_and_deletes_folder(self, tmp_path: Path) -> None:
|
|
old_path = tmp_path / "old_orphan"
|
|
old_path.mkdir()
|
|
new_path = tmp_path / "new"
|
|
new_path.mkdir()
|
|
file1 = old_path / "S01E01.mkv"
|
|
file1.write_text("episode 1")
|
|
file2 = old_path / "S01E02.mkv"
|
|
file2.write_text("episode 2")
|
|
result = _cleanup_orphaned_folder(old_path, new_path)
|
|
assert result is True
|
|
assert not old_path.exists()
|
|
assert (new_path / "S01E01.mkv").exists()
|
|
assert (new_path / "S01E02.mkv").exists()
|
|
|
|
def test_dry_run_does_not_delete_empty_folder(self, tmp_path: Path) -> None:
|
|
old_path = tmp_path / "empty_orphan"
|
|
old_path.mkdir()
|
|
new_path = tmp_path / "new"
|
|
new_path.mkdir()
|
|
result = _cleanup_orphaned_folder(old_path, new_path, dry_run=True)
|
|
assert result is True
|
|
assert old_path.exists()
|
|
|
|
def test_dry_run_does_not_move_files(self, tmp_path: Path) -> None:
|
|
old_path = tmp_path / "old_orphan"
|
|
old_path.mkdir()
|
|
new_path = tmp_path / "new"
|
|
new_path.mkdir()
|
|
file1 = old_path / "S01E01.mkv"
|
|
file1.write_text("episode 1")
|
|
result = _cleanup_orphaned_folder(old_path, new_path, dry_run=True)
|
|
assert result is True
|
|
assert old_path.exists()
|
|
assert not (new_path / "S01E01.mkv").exists()
|
|
|
|
def test_handles_permission_error_gracefully(self, tmp_path: Path) -> None:
|
|
old_path = tmp_path / "permission_denied"
|
|
old_path.mkdir()
|
|
new_path = tmp_path / "new"
|
|
new_path.mkdir()
|
|
# Simulate permission error by patching rmdir
|
|
with patch.object(Path, "rmdir", side_effect=PermissionError("Access denied")):
|
|
result = _cleanup_orphaned_folder(old_path, new_path)
|
|
assert result is False
|
|
|
|
|
|
class TestValidateAndRenameSeriesFolders:
|
|
"""Integration-style tests for validate_and_rename_series_folders."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_no_anime_directory(self) -> None:
|
|
with patch(
|
|
"src.server.services.folder_rename_service.settings.anime_directory",
|
|
"",
|
|
):
|
|
stats = await validate_and_rename_series_folders()
|
|
assert stats == {"scanned": 0, "renamed": 0, "skipped": 0, "errors": 0}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_renames_folder_when_name_differs(self, tmp_path: Path) -> None:
|
|
anime_dir = tmp_path / "anime"
|
|
anime_dir.mkdir()
|
|
series_dir = anime_dir / "Attack on Titan"
|
|
series_dir.mkdir()
|
|
(series_dir / "tvshow.nfo").write_text(
|
|
"<tvshow><title>Attack on Titan</title><year>2013</year></tvshow>"
|
|
)
|
|
|
|
with patch(
|
|
"src.server.services.folder_rename_service.settings.anime_directory",
|
|
str(anime_dir),
|
|
), patch(
|
|
"src.server.services.folder_rename_service._is_series_being_downloaded",
|
|
return_value=False,
|
|
), patch(
|
|
"src.server.services.folder_rename_service._update_database_paths",
|
|
new_callable=AsyncMock,
|
|
) as mock_update_db:
|
|
stats = await validate_and_rename_series_folders()
|
|
|
|
assert stats["scanned"] == 1
|
|
assert stats["renamed"] == 1
|
|
assert stats["skipped"] == 0
|
|
assert stats["errors"] == 0
|
|
assert not series_dir.exists()
|
|
assert (anime_dir / "Attack on Titan (2013)").is_dir()
|
|
mock_update_db.assert_awaited_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_skips_when_name_already_correct(self, tmp_path: Path) -> None:
|
|
anime_dir = tmp_path / "anime"
|
|
anime_dir.mkdir()
|
|
series_dir = anime_dir / "Attack on Titan (2013)"
|
|
series_dir.mkdir()
|
|
(series_dir / "tvshow.nfo").write_text(
|
|
"<tvshow><title>Attack on Titan</title><year>2013</year></tvshow>"
|
|
)
|
|
|
|
with patch(
|
|
"src.server.services.folder_rename_service.settings.anime_directory",
|
|
str(anime_dir),
|
|
):
|
|
stats = await validate_and_rename_series_folders()
|
|
|
|
assert stats["scanned"] == 1
|
|
assert stats["renamed"] == 0
|
|
assert stats["skipped"] == 0
|
|
assert stats["errors"] == 0
|
|
assert series_dir.is_dir()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_skips_missing_title_or_year(self, tmp_path: Path) -> None:
|
|
anime_dir = tmp_path / "anime"
|
|
anime_dir.mkdir()
|
|
series_dir = anime_dir / "Incomplete"
|
|
series_dir.mkdir()
|
|
(series_dir / "tvshow.nfo").write_text(
|
|
"<tvshow><title>Incomplete</title></tvshow>"
|
|
)
|
|
|
|
with patch(
|
|
"src.server.services.folder_rename_service.settings.anime_directory",
|
|
str(anime_dir),
|
|
):
|
|
stats = await validate_and_rename_series_folders()
|
|
|
|
assert stats["scanned"] == 1
|
|
assert stats["renamed"] == 0
|
|
assert stats["skipped"] == 1
|
|
assert stats["errors"] == 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_skips_when_download_active(self, tmp_path: Path) -> None:
|
|
anime_dir = tmp_path / "anime"
|
|
anime_dir.mkdir()
|
|
series_dir = anime_dir / "Attack on Titan"
|
|
series_dir.mkdir()
|
|
(series_dir / "tvshow.nfo").write_text(
|
|
"<tvshow><title>Attack on Titan</title><year>2013</year></tvshow>"
|
|
)
|
|
|
|
with patch(
|
|
"src.server.services.folder_rename_service.settings.anime_directory",
|
|
str(anime_dir),
|
|
), patch(
|
|
"src.server.services.folder_rename_service._is_series_being_downloaded",
|
|
return_value=True,
|
|
):
|
|
stats = await validate_and_rename_series_folders()
|
|
|
|
assert stats["scanned"] == 1
|
|
assert stats["renamed"] == 0
|
|
assert stats["skipped"] == 1
|
|
assert stats["errors"] == 0
|
|
assert series_dir.is_dir()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_errors_when_target_exists(self, tmp_path: Path) -> None:
|
|
anime_dir = tmp_path / "anime"
|
|
anime_dir.mkdir()
|
|
series_dir = anime_dir / "Attack on Titan"
|
|
series_dir.mkdir()
|
|
(series_dir / "tvshow.nfo").write_text(
|
|
"<tvshow><title>Attack on Titan</title><year>2013</year></tvshow>"
|
|
)
|
|
# Pre-create the target folder to simulate a duplicate
|
|
(anime_dir / "Attack on Titan (2013)").mkdir()
|
|
|
|
with patch(
|
|
"src.server.services.folder_rename_service.settings.anime_directory",
|
|
str(anime_dir),
|
|
), patch(
|
|
"src.server.services.folder_rename_service._is_series_being_downloaded",
|
|
return_value=False,
|
|
):
|
|
stats = await validate_and_rename_series_folders()
|
|
|
|
assert stats["scanned"] == 1
|
|
assert stats["renamed"] == 0
|
|
assert stats["skipped"] == 0
|
|
assert stats["errors"] == 1
|
|
assert series_dir.is_dir()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_counts_multiple_folders(self, tmp_path: Path) -> None:
|
|
anime_dir = tmp_path / "anime"
|
|
anime_dir.mkdir()
|
|
|
|
# Folder 1: needs rename
|
|
d1 = anime_dir / "Show A"
|
|
d1.mkdir()
|
|
(d1 / "tvshow.nfo").write_text(
|
|
"<tvshow><title>Show A</title><year>2020</year></tvshow>"
|
|
)
|
|
|
|
# Folder 2: already correct
|
|
d2 = anime_dir / "Show B (2021)"
|
|
d2.mkdir()
|
|
(d2 / "tvshow.nfo").write_text(
|
|
"<tvshow><title>Show B</title><year>2021</year></tvshow>"
|
|
)
|
|
|
|
# Folder 3: missing year
|
|
d3 = anime_dir / "Show C"
|
|
d3.mkdir()
|
|
(d3 / "tvshow.nfo").write_text("<tvshow><title>Show C</title></tvshow>")
|
|
|
|
with patch(
|
|
"src.server.services.folder_rename_service.settings.anime_directory",
|
|
str(anime_dir),
|
|
), patch(
|
|
"src.server.services.folder_rename_service._is_series_being_downloaded",
|
|
return_value=False,
|
|
), patch(
|
|
"src.server.services.folder_rename_service._update_database_paths",
|
|
new_callable=AsyncMock,
|
|
):
|
|
stats = await validate_and_rename_series_folders()
|
|
|
|
assert stats["scanned"] == 3
|
|
assert stats["renamed"] == 1
|
|
assert stats["skipped"] == 1
|
|
assert stats["errors"] == 0
|
|
assert not d1.exists()
|
|
assert (anime_dir / "Show A (2020)").is_dir()
|
|
assert d2.is_dir()
|
|
assert d3.is_dir()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dry_run_does_not_rename_folders(self, tmp_path: Path) -> None:
|
|
anime_dir = tmp_path / "anime"
|
|
anime_dir.mkdir()
|
|
series_dir = anime_dir / "Attack on Titan"
|
|
series_dir.mkdir()
|
|
(series_dir / "tvshow.nfo").write_text(
|
|
"<tvshow><title>Attack on Titan</title><year>2013</year></tvshow>"
|
|
)
|
|
|
|
with patch(
|
|
"src.server.services.folder_rename_service.settings.anime_directory",
|
|
str(anime_dir),
|
|
), patch(
|
|
"src.server.services.folder_rename_service._is_series_being_downloaded",
|
|
return_value=False,
|
|
):
|
|
stats = await validate_and_rename_series_folders(dry_run=True)
|
|
|
|
assert stats["scanned"] == 1
|
|
assert stats["renamed"] == 1
|
|
assert stats["skipped"] == 0
|
|
assert stats["errors"] == 0
|
|
# Original folder should still exist (not renamed in dry-run)
|
|
assert series_dir.is_dir()
|
|
assert not (anime_dir / "Attack on Titan (2013)").exists()
|