"""Tests for the smaller ``alfred.infrastructure.filesystem`` helpers. Covers three siblings of ``FileManager`` that had near-zero coverage: - ``filesystem_operations.create_folder`` / ``move`` — thin ``mkdir`` / ``mv`` wrappers returning dict-shaped responses. - ``organizer.MediaOrganizer`` — computes destination paths for movies and TV episodes; creates folders for them. - ``find_video.find_video_file`` — first-video lookup in a folder. (``ffprobe`` coverage now lives in ``test_ffprobe_prober.py`` alongside its adapter.) """ from __future__ import annotations from unittest.mock import MagicMock, patch from alfred.domain.movies.entities import Movie from alfred.domain.movies.value_objects import MovieTitle, Quality, ReleaseYear from alfred.domain.shared.value_objects import ImdbId from alfred.domain.tv_shows.entities import Episode, TVShow from alfred.domain.tv_shows.value_objects import ( EpisodeNumber, SeasonNumber, ShowStatus, ) from alfred.infrastructure.filesystem.filesystem_operations import ( create_folder, move, ) from alfred.infrastructure.filesystem.find_video import find_video_file from alfred.infrastructure.filesystem.organizer import MediaOrganizer from alfred.infrastructure.knowledge.release_kb import YamlReleaseKnowledge _KB = YamlReleaseKnowledge() # --------------------------------------------------------------------------- # # filesystem_operations # # --------------------------------------------------------------------------- # class TestCreateFolder: def test_creates_nested(self, tmp_path): target = tmp_path / "a" / "b" / "c" out = create_folder(str(target)) assert out == {"status": "ok", "path": str(target)} assert target.is_dir() def test_existing_is_ok(self, tmp_path): out = create_folder(str(tmp_path)) assert out["status"] == "ok" def test_os_error_wrapped(self, tmp_path): with patch( "alfred.infrastructure.filesystem.filesystem_operations.Path.mkdir", side_effect=OSError("readonly fs"), ): out = create_folder(str(tmp_path / "x")) assert out == { "status": "error", "error": "mkdir_failed", "message": "readonly fs", } class TestMove: def test_source_not_found(self, tmp_path): out = move(str(tmp_path / "ghost"), str(tmp_path / "dst")) assert out["status"] == "error" assert out["error"] == "source_not_found" def test_destination_exists(self, tmp_path): src = tmp_path / "src" src.write_text("x") dst = tmp_path / "dst" dst.write_text("y") out = move(str(src), str(dst)) assert out["error"] == "destination_exists" def test_happy_path_returns_ok(self, tmp_path): src = tmp_path / "src" src.write_text("x") dst = tmp_path / "dst" # Patch subprocess so we don't actually shell out; pretend success. with patch( "alfred.infrastructure.filesystem.filesystem_operations.subprocess.run", return_value=MagicMock(returncode=0, stderr=""), ): out = move(str(src), str(dst)) assert out == {"status": "ok", "source": str(src), "destination": str(dst)} def test_mv_failure_wrapped(self, tmp_path): src = tmp_path / "src" src.write_text("x") with patch( "alfred.infrastructure.filesystem.filesystem_operations.subprocess.run", return_value=MagicMock(returncode=1, stderr="cross-device link\n"), ): out = move(str(src), str(tmp_path / "dst")) assert out["error"] == "move_failed" assert out["message"] == "cross-device link" def test_os_error_wrapped(self, tmp_path): src = tmp_path / "src" src.write_text("x") with patch( "alfred.infrastructure.filesystem.filesystem_operations.subprocess.run", side_effect=OSError("ENOSPC"), ): out = move(str(src), str(tmp_path / "dst")) assert out["error"] == "move_failed" # --------------------------------------------------------------------------- # # find_video # # --------------------------------------------------------------------------- # class TestFindVideo: def test_returns_file_directly_when_video(self, tmp_path): f = tmp_path / "Movie.mkv" f.write_bytes(b"") assert find_video_file(f, _KB) == f def test_returns_none_when_file_is_not_video(self, tmp_path): f = tmp_path / "notes.txt" f.write_text("x") assert find_video_file(f, _KB) is None def test_returns_none_when_folder_has_no_video(self, tmp_path): (tmp_path / "a.txt").write_text("x") assert find_video_file(tmp_path, _KB) is None def test_returns_first_sorted_video(self, tmp_path): (tmp_path / "B.mkv").write_bytes(b"") (tmp_path / "A.mkv").write_bytes(b"") (tmp_path / "C.mkv").write_bytes(b"") found = find_video_file(tmp_path, _KB) assert found.name == "A.mkv" def test_recurses_into_subfolders(self, tmp_path): sub = tmp_path / "sub" sub.mkdir() (sub / "X.mkv").write_bytes(b"") found = find_video_file(tmp_path, _KB) assert found is not None and found.name == "X.mkv" def test_case_insensitive_extension(self, tmp_path): f = tmp_path / "Movie.MKV" f.write_bytes(b"") assert find_video_file(f, _KB) == f # --------------------------------------------------------------------------- # # MediaOrganizer # # --------------------------------------------------------------------------- # def _movie() -> Movie: return Movie( imdb_id=ImdbId("tt1375666"), title=MovieTitle("Inception"), release_year=ReleaseYear(2010), quality=Quality.HD, ) def _show() -> TVShow: return TVShow( imdb_id=ImdbId("tt0773262"), title="Dexter", expected_seasons=8, status=ShowStatus.ENDED, ) def _episode() -> Episode: return Episode( season_number=SeasonNumber(1), episode_number=EpisodeNumber(1), title="Dexter", ) class TestMediaOrganizer: def test_get_movie_destination(self, tmp_path): org = MediaOrganizer(tmp_path / "movies", tmp_path / "tv") out = org.get_movie_destination(_movie(), "source.mkv") # Path: /movies//.mkv assert out.suffix == ".mkv" assert out.parent.name == _movie().get_folder_name() assert out.parent.parent == tmp_path / "movies" def test_get_movie_destination_preserves_extension(self, tmp_path): org = MediaOrganizer(tmp_path / "movies", tmp_path / "tv") out = org.get_movie_destination(_movie(), "source.MP4") assert out.suffix == ".MP4" def test_get_episode_destination(self, tmp_path): org = MediaOrganizer(tmp_path / "movies", tmp_path / "tv") out = org.get_episode_destination(_show(), _episode(), "raw.mkv") # Path: /tv///.mkv assert out.suffix == ".mkv" assert out.parent.parent.parent == tmp_path / "tv" assert out.parent.parent.name == _show().get_folder_name() def test_create_movie_directory_creates_folder(self, tmp_path): org = MediaOrganizer(tmp_path / "movies", tmp_path / "tv") assert org.create_movie_directory(_movie()) is True assert (tmp_path / "movies" / _movie().get_folder_name()).is_dir() def test_create_movie_directory_already_exists_ok(self, tmp_path): org = MediaOrganizer(tmp_path / "movies", tmp_path / "tv") org.create_movie_directory(_movie()) # Second call is also fine (parents=True, exist_ok=True). assert org.create_movie_directory(_movie()) is True def test_create_movie_directory_failure_returns_false(self, tmp_path): org = MediaOrganizer(tmp_path / "movies", tmp_path / "tv") with patch( "alfred.infrastructure.filesystem.organizer.Path.mkdir", side_effect=PermissionError("denied"), ): assert org.create_movie_directory(_movie()) is False def test_create_episode_directory_creates_season_folder(self, tmp_path): org = MediaOrganizer(tmp_path / "movies", tmp_path / "tv") assert org.create_episode_directory(_show(), 1) is True # /tv// exists show_dir = tmp_path / "tv" / _show().get_folder_name() assert show_dir.is_dir() # At least one child (the season folder) was created. assert any(show_dir.iterdir()) def test_create_episode_directory_failure_returns_false(self, tmp_path): org = MediaOrganizer(tmp_path / "movies", tmp_path / "tv") with patch( "alfred.infrastructure.filesystem.organizer.Path.mkdir", side_effect=OSError("readonly"), ): assert org.create_episode_directory(_show(), 1) is False