immich-extern-to-album/tests/test_main.py

import importlib.util
import unittest
from pathlib import Path
from unittest.mock import patch


MODULE_PATH = Path(__file__).parents[1] / "src" / "main.py"
SPEC = importlib.util.spec_from_file_location("album_sync", MODULE_PATH)
assert SPEC and SPEC.loader
album_sync = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(album_sync)


class AlbumNameTests(unittest.TestCase):
    def test_root_directory_becomes_album_name(self) -> None:
        with patch.object(
            album_sync,
            "EXTERNAL_ROOT",
            "/mnt/nextcloud-photos",
        ), patch.object(album_sync, "ALBUM_MODE", "root"):
            self.assertEqual(
                album_sync.get_album_name(
                    "/mnt/nextcloud-photos/Korea_2016_07_08/photo.jpg"
                ),
                "Korea_2016_07_08",
            )

    def test_nested_files_stay_in_root_album(self) -> None:
        with patch.object(
            album_sync,
            "EXTERNAL_ROOT",
            "/mnt/nextcloud-photos",
        ), patch.object(album_sync, "ALBUM_MODE", "root"):
            self.assertEqual(
                album_sync.get_album_name(
                    "/mnt/nextcloud-photos/Korea_2016_07_08/day-1/photo.jpg"
                ),
                "Korea_2016_07_08",
            )

    def test_file_directly_in_external_root_is_skipped(self) -> None:
        with patch.object(
            album_sync,
            "EXTERNAL_ROOT",
            "/mnt/nextcloud-photos",
        ):
            self.assertIsNone(
                album_sync.get_album_name(
                    "/mnt/nextcloud-photos/photo.jpg"
                )
            )

    def test_similar_root_prefix_is_skipped(self) -> None:
        with patch.object(
            album_sync,
            "EXTERNAL_ROOT",
            "/mnt/nextcloud-photos",
        ):
            self.assertIsNone(
                album_sync.get_album_name(
                    "/mnt/nextcloud-photos-old/Korea/photo.jpg"
                )
            )

    def test_parent_traversal_cannot_escape_external_root(self) -> None:
        with patch.object(
            album_sync,
            "EXTERNAL_ROOT",
            "/mnt/nextcloud-photos",
        ):
            self.assertIsNone(
                album_sync.get_album_name(
                    "/mnt/nextcloud-photos/../other/Korea/photo.jpg"
                )
            )


class SearchTests(unittest.TestCase):
    def test_search_assets_follows_pagination(self) -> None:
        responses = [
            {
                "assets": {
                    "items": [{"id": "one"}],
                    "nextPage": "2",
                }
            },
            {
                "assets": {
                    "items": [{"id": "two"}],
                    "nextPage": None,
                }
            },
        ]

        with patch.object(
            album_sync,
            "api_request",
            side_effect=responses,
        ) as request:
            assets = album_sync.search_assets(libraryId="library-id")

        self.assertEqual(assets, [{"id": "one"}, {"id": "two"}])
        self.assertEqual(request.call_count, 2)
        self.assertEqual(
            request.call_args_list[0].kwargs["json"],
            {
                "libraryId": "library-id",
                "page": 1,
                "size": album_sync.PAGE_SIZE,
            },
        )


class SynchronizationTests(unittest.TestCase):
    def test_main_creates_folder_album_with_all_assets(self) -> None:
        assets = [
            {
                "id": "asset-one",
                "originalPath": (
                    "/mnt/nextcloud-photos/"
                    "Korea_2016_07_08/photo-1.jpg"
                ),
            },
            {
                "id": "asset-two",
                "originalPath": (
                    "/mnt/nextcloud-photos/"
                    "Korea_2016_07_08/day-2/photo-2.jpg"
                ),
            },
        ]
        created_album = {
            "id": "album-id",
            "albumName": "Korea_2016_07_08",
        }

        with patch.object(
            album_sync,
            "EXTERNAL_ROOT",
            "/mnt/nextcloud-photos",
        ), patch.object(
            album_sync,
            "ALBUM_MODE",
            "root",
        ), patch.object(
            album_sync,
            "SYNC_INTERVAL_SECONDS",
            0,
        ), patch.object(
            album_sync,
            "check_configuration",
        ), patch.object(
            album_sync,
            "get_albums",
            return_value={},
        ), patch.object(
            album_sync,
            "get_external_assets",
            return_value=assets,
        ), patch.object(
            album_sync,
            "create_album",
            return_value=created_album,
        ) as create_album, patch.object(
            album_sync,
            "get_album_asset_ids",
            return_value=set(),
        ), patch.object(
            album_sync,
            "add_assets_to_album",
        ) as add_assets:
            result = album_sync.main()

        self.assertEqual(result, 0)
        create_album.assert_called_once_with("Korea_2016_07_08")
        add_assets.assert_called_once_with(
            "album-id",
            ["asset-one", "asset-two"],
        )

    def test_continuous_mode_waits_and_runs_again(self) -> None:
        with patch.object(
            album_sync,
            "SYNC_INTERVAL_SECONDS",
            1800,
        ), patch.object(
            album_sync,
            "check_configuration",
        ), patch.object(
            album_sync,
            "synchronize",
            side_effect=[None, KeyboardInterrupt],
        ) as synchronize, patch.object(
            album_sync.time,
            "sleep",
        ) as sleep:
            with self.assertRaises(KeyboardInterrupt):
                album_sync.main()

        self.assertEqual(synchronize.call_count, 2)
        sleep.assert_called_once_with(1800)

    def test_continuous_mode_retries_after_sync_error(self) -> None:
        with patch.object(
            album_sync,
            "SYNC_INTERVAL_SECONDS",
            1800,
        ), patch.object(
            album_sync,
            "check_configuration",
        ), patch.object(
            album_sync,
            "synchronize",
            side_effect=[RuntimeError("temporary"), KeyboardInterrupt],
        ) as synchronize, patch.object(
            album_sync.time,
            "sleep",
        ) as sleep:
            with self.assertRaises(KeyboardInterrupt):
                album_sync.main()

        self.assertEqual(synchronize.call_count, 2)
        sleep.assert_called_once_with(1800)


if __name__ == "__main__":
    unittest.main()