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]