tg-favorite-stickers/tests/test_service.py

from pathlib import Path

import pytest

from favorite_stickers.media import ConvertedMedia, IncomingMedia
from favorite_stickers.service import (
    FavoriteStickerService,
    RemoteSticker,
    ToggleAction,
)
from favorite_stickers.store import Store


class FakeConverter:
    def __init__(self) -> None:
        self.calls = 0

    async def convert(self, media: IncomingMedia) -> ConvertedMedia:
        self.calls += 1
        return ConvertedMedia(Path("result.webm"), media.content_hash or "hash")


class FakeTelegram:
    def __init__(self) -> None:
        self.created: list[str] = []
        self.added: list[str] = []
        self.deleted: list[str] = []
        self.packs: dict[str, list[RemoteSticker]] = {}
        self.number = 0

    async def create_pack(
        self,
        user_id: int,
        name: str,
        title: str,
        sticker_path: Path,
        emoji: str,
    ) -> RemoteSticker:
        self.created.append(name)
        remote = self._remote()
        self.packs[name] = [remote]
        return remote

    async def add_sticker(
        self, user_id: int, name: str, sticker_path: Path, emoji: str
    ) -> RemoteSticker:
        self.added.append(name)
        remote = self._remote()
        self.packs[name].append(remote)
        return remote

    async def delete_sticker(self, sticker_file_id: str) -> None:
        self.deleted.append(sticker_file_id)
        for stickers in self.packs.values():
            stickers[:] = [item for item in stickers if item.file_id != sticker_file_id]

    async def get_pack(self, name: str) -> list[RemoteSticker] | None:
        stickers = self.packs.get(name)
        return list(stickers) if stickers is not None else None

    def _remote(self) -> RemoteSticker:
        self.number += 1
        return RemoteSticker(
            file_id=f"file-{self.number}", unique_id=f"unique-{self.number}"
        )


@pytest.mark.asyncio
async def test_first_media_creates_video_pack_without_questions(tmp_path) -> None:
    store = Store(tmp_path / "bot.sqlite3")
    store.initialize()
    telegram = FakeTelegram()
    converter = FakeConverter()
    service = FavoriteStickerService(store, telegram, converter, "TestBot")

    result = await service.toggle(
        7, IncomingMedia(Path("source.png"), "source-id", "static", "❤️", "hash")
    )

    assert result.action is ToggleAction.ADDED
    assert result.pack_name == "favorites_7_by_testbot"
    assert telegram.created == [result.pack_name]
    pack = store.latest_pack(7)
    assert pack is not None
    assert pack.sticker_count == 1


@pytest.mark.asyncio
async def test_resending_original_sticker_removes_it_without_conversion(
    tmp_path,
) -> None:
    store = Store(tmp_path / "bot.sqlite3")
    store.initialize()
    telegram = FakeTelegram()
    converter = FakeConverter()
    service = FavoriteStickerService(store, telegram, converter, "TestBot")
    media = IncomingMedia(Path("source.webp"), "same-id", "static", "👍", "same-hash")
    await service.toggle(7, media)

    result = await service.toggle(7, media)

    assert result.action is ToggleAction.REMOVED
    assert telegram.deleted == ["file-1"]
    assert converter.calls == 1
    pack = store.latest_pack(7)
    assert pack is not None
    assert pack.sticker_count == 0


@pytest.mark.asyncio
async def test_full_pack_rolls_over_to_numbered_pack(tmp_path) -> None:
    store = Store(tmp_path / "bot.sqlite3")
    store.initialize()
    telegram = FakeTelegram()
    converter = FakeConverter()
    service = FavoriteStickerService(
        store, telegram, converter, "TestBot", pack_limit=2
    )

    results = []
    for index in range(3):
        results.append(
            await service.toggle(
                7,
                IncomingMedia(
                    Path(f"{index}.png"),
                    f"source-{index}",
                    "static",
                    "❤️",
                    f"hash-{index}",
                ),
            )
        )

    assert telegram.created == [
        "favorites_7_by_testbot",
        "favorites_7_2_by_testbot",
    ]
    assert telegram.added == ["favorites_7_by_testbot"]
    assert results[-1].pack_name == "favorites_7_2_by_testbot"


class FailOnceStore(Store):
    fail_add = False
    fail_remove = False

    def complete_add(self, user_id: int, file_id: str, unique_id: str) -> None:
        if self.fail_add:
            self.fail_add = False
            raise RuntimeError("simulated SQLite add failure")
        super().complete_add(user_id, file_id, unique_id)

    def complete_remove(self, user_id: int) -> None:
        if self.fail_remove:
            self.fail_remove = False
            raise RuntimeError("simulated SQLite remove failure")
        super().complete_remove(user_id)


@pytest.mark.asyncio
async def test_pending_add_is_recovered_after_sqlite_failure(tmp_path) -> None:
    store = FailOnceStore(tmp_path / "bot.sqlite3")
    store.initialize()
    store.fail_add = True
    telegram = FakeTelegram()
    converter = FakeConverter()
    service = FavoriteStickerService(store, telegram, converter, "TestBot")
    media = IncomingMedia(Path("source.png"), "source", "static", "❤️", "hash")

    with pytest.raises(RuntimeError, match="SQLite add"):
        await service.toggle(7, media)

    assert store.pending_operation(7) is not None
    result = await service.toggle(7, media)
    assert store.pending_operation(7) is None
    assert result.action is ToggleAction.ADDED
    assert telegram.deleted == []
    assert converter.calls == 1
    assert store.find_sticker(7, telegram_unique_id="source") is not None


@pytest.mark.asyncio
async def test_pending_remove_is_recovered_after_sqlite_failure(tmp_path) -> None:
    store = FailOnceStore(tmp_path / "bot.sqlite3")
    store.initialize()
    telegram = FakeTelegram()
    converter = FakeConverter()
    service = FavoriteStickerService(store, telegram, converter, "TestBot")
    media = IncomingMedia(Path("source.png"), "source", "static", "❤️", "hash")
    await service.toggle(7, media)
    store.fail_remove = True

    with pytest.raises(RuntimeError, match="SQLite remove"):
        await service.toggle(7, media)

    assert store.pending_operation(7) is not None
    result = await service.toggle(7, media)
    assert store.pending_operation(7) is None
    assert result.action is ToggleAction.REMOVED
    assert telegram.created == ["favorites_7_by_testbot"]
    assert telegram.added == []
    assert converter.calls == 1
    assert store.find_sticker(7, telegram_unique_id="source") is None