docker-images/linkki-tiedotus/tests/test_workflow.py

from pathlib import Path

from linkki_poster.models import Config, Language, LanguageAssets
from linkki_poster.broadcast_ops import order_scan_languages
from linkki_poster.models import ScanResult
from linkki_poster.telegram_api import TelegramApiError
from linkki_poster.workflow import (
    extract_internal_chat_id,
    post_assets,
    resolve_link_target,
    split_markdown_text_for_messages,
)


class FakeTelegramAPI:
    def __init__(self, fail_caption_too_long: bool = False) -> None:
        self.next_message_id = 100
        self.calls: list[tuple[str, dict]] = []
        self.fail_caption_too_long = fail_caption_too_long

    def send_media_group(
        self,
        chat_id: str,
        image_paths: list[Path],
        caption: str | None = None,
        parse_mode: str | None = None,
    ) -> list[dict]:
        self.calls.append(
            (
                "send_media_group",
                {
                    "chat_id": chat_id,
                    "images": [path.name for path in image_paths],
                    "caption": caption,
                    "parse_mode": parse_mode,
                },
            )
        )
        if caption is not None and self.fail_caption_too_long:
            raise TelegramApiError("Bad Request: caption is too long")
        base = self.next_message_id
        self.next_message_id += len(image_paths)
        return [{"message_id": base + index} for index, _ in enumerate(image_paths)]

    def send_message(self, chat_id: str, text: str, parse_mode: str | None = None) -> dict:
        self.calls.append(
            (
                "send_message",
                {
                    "chat_id": chat_id,
                    "text": text,
                    "parse_mode": parse_mode,
                },
            )
        )
        message_id = self.next_message_id
        self.next_message_id += 1
        return {"message_id": message_id}

    def edit_message_text(self, chat_id: str, message_id: int, text: str, parse_mode: str | None = None) -> dict:
        self.calls.append(
            (
                "edit_message_text",
                {
                    "chat_id": chat_id,
                    "message_id": message_id,
                    "text": text,
                    "parse_mode": parse_mode,
                },
            )
        )
        return {"message_id": message_id}

    def edit_message_caption(
        self,
        chat_id: str,
        message_id: int,
        caption: str,
        parse_mode: str | None = None,
    ) -> dict:
        self.calls.append(
            (
                "edit_message_caption",
                {
                    "chat_id": chat_id,
                    "message_id": message_id,
                    "caption": caption,
                    "parse_mode": parse_mode,
                },
            )
        )
        return {"message_id": message_id}

    def delete_message(self, chat_id: str, message_id: int) -> bool:
        self.calls.append(
            (
                "delete_message",
                {
                    "chat_id": chat_id,
                    "message_id": message_id,
                },
            )
        )
        return True


def test_post_assets_attaches_globals_to_first_language_without_images(tmp_path: Path) -> None:
    config = Config(
        bot_token="token",
        chat_id="-100123",
        channel_username="",
    )
    fi = LanguageAssets(
        language=Language.FI,
        text_file=tmp_path / "Finnish.md",
        text_raw="Read [English](en)",
        images=[],
    )
    en = LanguageAssets(
        language=Language.EN,
        text_file=tmp_path / "English.md",
        text_raw="Hello",
        images=[tmp_path / "English1.jpg"],
    )
    global_images = [tmp_path / "bild1.jpg", tmp_path / "bild2.jpg"]

    api = FakeTelegramAPI()
    refs = post_assets(api=api, config=config, global_images=global_images, ordered_languages=[fi, en])

    assert api.calls[0][0] == "send_media_group"
    assert api.calls[0][1]["images"] == ["bild1.jpg", "bild2.jpg"]
    assert api.calls[0][1]["caption"] is not None

    send_message_calls = [call for call in api.calls if call[0] == "send_message"]
    assert len(send_message_calls) == 0

    edit_caption_calls = [call for call in api.calls if call[0] == "edit_message_caption"]
    assert len(edit_caption_calls) >= 1
    assert 'href="https://t.me/c/123/' in edit_caption_calls[-1][1]["caption"]

    assert len(refs) == 2
    assert refs[0].language == Language.FI
    assert refs[1].language == Language.EN
    assert refs[0].text_mode == "caption"


def test_post_assets_posts_globals_first_when_first_language_has_images(tmp_path: Path) -> None:
    config = Config(bot_token="token", chat_id="-100123", channel_username="")
    fi = LanguageAssets(
        language=Language.FI,
        text_file=tmp_path / "Finnish.md",
        text_raw="Hei",
        images=[tmp_path / "suomi1.jpg"],
    )
    en = LanguageAssets(
        language=Language.EN,
        text_file=tmp_path / "English.md",
        text_raw="Hello",
        images=[],
    )
    global_images = [tmp_path / "bild1.jpg", tmp_path / "bild2.jpg"]

    api = FakeTelegramAPI()
    post_assets(api=api, config=config, global_images=global_images, ordered_languages=[fi, en])

    assert api.calls[0][0] == "send_media_group"
    assert api.calls[0][1]["images"] == ["bild1.jpg", "bild2.jpg"]
    assert api.calls[0][1]["caption"] is None


def test_long_text_with_images_is_not_caption(tmp_path: Path) -> None:
    config = Config(bot_token="token", chat_id="-100123", channel_username="")
    long_text = "x" * 1200
    en = LanguageAssets(
        language=Language.EN,
        text_file=tmp_path / "English.md",
        text_raw=long_text,
        images=[tmp_path / "english1.jpg"],
    )

    api = FakeTelegramAPI()
    refs = post_assets(api=api, config=config, global_images=[], ordered_languages=[en])

    first_call = api.calls[0]
    assert first_call[0] == "send_media_group"
    assert first_call[1]["caption"] is None
    second_call = api.calls[1]
    assert second_call[0] == "send_message"
    assert refs[0].text_mode == "message"


def test_caption_too_long_error_falls_back_to_media_then_message(tmp_path: Path) -> None:
    config = Config(bot_token="token", chat_id="-100123", channel_username="")
    en = LanguageAssets(
        language=Language.EN,
        text_file=tmp_path / "English.md",
        text_raw="short text",
        images=[tmp_path / "english1.jpg"],
    )

    api = FakeTelegramAPI(fail_caption_too_long=True)
    refs = post_assets(api=api, config=config, global_images=[], ordered_languages=[en])

    assert api.calls[0][0] == "send_media_group"
    assert api.calls[0][1]["caption"] == "short text"
    assert api.calls[1][0] == "send_media_group"
    assert api.calls[1][1]["caption"] is None
    assert api.calls[2][0] == "send_message"
    assert refs[0].text_mode == "message"


def test_split_prefers_double_newline_boundaries() -> None:
    text = ("a" * 2500) + "\n\n" + ("b" * 2500)
    chunks = split_markdown_text_for_messages(text, link_target="chan", message_limit=4096)

    assert len(chunks) == 2
    assert chunks[0].endswith("\n\n")
    assert chunks[1].startswith("b")


def test_split_does_not_force_extra_paragraph_split_when_mid_split_is_needed() -> None:
    text = ("x" * 5000) + "\n\n" + ("tail" * 10)
    chunks = split_markdown_text_for_messages(text, link_target="chan", message_limit=4096)

    assert len(chunks) == 2
    assert chunks[0].endswith("x")


def test_resolve_link_target_supports_private_channel_style() -> None:
    assert resolve_link_target("-1003896755764", "") == "c/3896755764"
    assert extract_internal_chat_id("-1003896755764") == "3896755764"


def test_resolve_link_target_prefers_public_username() -> None:
    assert resolve_link_target("-1003896755764", "mychannel") == "mychannel"


def test_order_scan_languages_respects_explicit_allowlist() -> None:
    scan_result = ScanResult(
        global_images=[],
        languages=[
            LanguageAssets(language=Language.FI, text_file=Path("Suomi.md"), text_raw="Hei", images=[]),
            LanguageAssets(language=Language.SV, text_file=Path("Svenska.md"), text_raw="Hej", images=[]),
            LanguageAssets(language=Language.EN, text_file=Path("English.md"), text_raw="Hello", images=[]),
        ],
    )

    ordered = order_scan_languages(scan_result, [Language.FI, Language.SV])

    assert [assets.language for assets in ordered] == [Language.FI, Language.SV]