Lukas
756731cd5d
- Remove NFO repair scan step from ARCHITECTURE.md startup sequence - Update CHANGELOG.md: rephrase perform_nfo_repair_scan as scheduled scan - Add test verifying perform_nfo_repair_scan is NOT called in lifespan - Keep existing folder scan wiring tests and unit tests intact - NFO_GUIDE.md already correctly describes scheduled scan behavior
384 lines
14 KiB
Python
384 lines
14 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 (
|
|
_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)"
|
|
|
|
|
|
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 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()
|