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

from linkki_poster.broadcast_ops import (
    DiscussionResolutionStats,
    apply_discussion_group_pinning,
    build_discussion_resolution_detail,
    prune_binding_after_deletions,
)
from linkki_poster.models import Config, Language, LanguageBinding, PublishBinding, TextSegmentBinding
from linkki_poster.telegram_api import TelegramApiError


def test_discussion_resolution_detail_mentions_missing_group_updates() -> None:
    detail = build_discussion_resolution_detail(
        DiscussionResolutionStats(
            updates_seen=4,
            discussion_group_updates=0,
            automatic_forward_updates=0,
            linked_channel_forward_updates=0,
        ),
        expected_count=2,
        matched_count=0,
    )

    assert "matched 0/2" in detail
    assert "none were from the configured discussion group" in detail


def test_discussion_resolution_detail_mentions_linked_channel_forwards() -> None:
    detail = build_discussion_resolution_detail(
        DiscussionResolutionStats(
            updates_seen=6,
            discussion_group_updates=5,
            automatic_forward_updates=3,
            linked_channel_forward_updates=2,
        ),
        expected_count=3,
        matched_count=2,
    )

    assert "matched 2/3" in detail
    assert "saw 2 automatic forwards from the linked channel" in detail


def test_discussion_resolution_detail_mentions_manual_linking_for_old_posts() -> None:
    detail = build_discussion_resolution_detail(
        DiscussionResolutionStats(
            updates_seen=0,
            discussion_group_updates=0,
            automatic_forward_updates=0,
            linked_channel_forward_updates=0,
        ),
        expected_count=3,
        matched_count=0,
    )

    assert "they must be linked manually" in detail


def test_apply_discussion_group_pinning_skips_service_message_cleanup_when_disabled(monkeypatch) -> None:
    calls = []

    class FakeTelegramAPI:
        def __init__(self, bot_token: str):
            self.bot_token = bot_token

        def pin_chat_message(self, chat_id: str, message_id: int, disable_notification: bool = True) -> bool:
            calls.append(("pin", chat_id, message_id, disable_notification))
            return True

        def unpin_chat_message(self, chat_id: str, message_id: int) -> bool:
            calls.append(("unpin", chat_id, message_id))
            return True

    def fail_if_delete_called(**kwargs):
        raise AssertionError("delete_pin_service_message should not be called")

    monkeypatch.setattr("linkki_poster.broadcast_ops.TelegramAPI", FakeTelegramAPI)
    monkeypatch.setattr("linkki_poster.broadcast_ops.delete_pin_service_message", fail_if_delete_called)

    binding = PublishBinding(
        global_message_ids=[],
        global_attached_to_language=None,
        languages={
            Language.FI: LanguageBinding(
                language=Language.FI,
                image_message_ids=[],
                text_segments=[TextSegmentBinding(message_id=101, mode="message")],
                primary_message_id=101,
                primary_message_url="https://t.me/c/123/101",
                discussion_message_id=201,
            ),
            Language.SV: LanguageBinding(
                language=Language.SV,
                image_message_ids=[],
                text_segments=[TextSegmentBinding(message_id=102, mode="message")],
                primary_message_id=102,
                primary_message_url="https://t.me/c/123/102",
                discussion_message_id=202,
            ),
        },
        discussion_message_ids={101: 201, 102: 202},
    )

    result = apply_discussion_group_pinning(
        Config(bot_token="token", chat_id="-1001", channel_username="channel", discussion_group_id="-2001"),
        binding,
        [Language.FI, Language.SV],
        "first",
        delete_service_messages=False,
    )

    assert result.binding == binding
    assert result.pinning_failed is False
    assert calls == [
        ("pin", "-2001", 201, True),
        ("unpin", "-2001", 202),
    ]


def test_prune_binding_after_deletions_keeps_remaining_channel_links() -> None:
    binding = PublishBinding(
        global_message_ids=[],
        global_attached_to_language=None,
        languages={
            Language.FI: LanguageBinding(
                language=Language.FI,
                image_message_ids=[],
                text_segments=[TextSegmentBinding(message_id=101, mode="message")],
                primary_message_id=101,
                primary_message_url="https://t.me/c/123/101",
                discussion_message_id=201,
            ),
            Language.SV: LanguageBinding(
                language=Language.SV,
                image_message_ids=[],
                text_segments=[TextSegmentBinding(message_id=102, mode="message")],
                primary_message_id=102,
                primary_message_url="https://t.me/c/123/102",
                discussion_message_id=202,
            ),
        },
        discussion_message_ids={101: 201, 102: 202},
    )

    remaining = prune_binding_after_deletions(
        binding,
        deleted_channel_ids=set(),
        deleted_discussion_channel_ids={101},
        deleted_discussion_message_ids={201},
        chat_id="-100123",
        channel_username="",
    )

    assert remaining is not None
    assert remaining.languages[Language.FI].text_segments[0].message_id == 101
    assert remaining.languages[Language.SV].text_segments[0].message_id == 102
    assert remaining.discussion_message_ids == {102: 202}


def test_apply_discussion_group_pinning_reports_permission_failures(monkeypatch) -> None:
    class FakeTelegramAPI:
        def __init__(self, bot_token: str):
            self.bot_token = bot_token

        def pin_chat_message(self, chat_id: str, message_id: int, disable_notification: bool = True) -> bool:
            raise TelegramApiError("Bad Request: not enough rights to manage pinned messages in the chat")

        def unpin_chat_message(self, chat_id: str, message_id: int) -> bool:
            raise TelegramApiError("Bad Request: not enough rights to manage pinned messages in the chat")

    monkeypatch.setattr("linkki_poster.broadcast_ops.TelegramAPI", FakeTelegramAPI)

    binding = PublishBinding(
        global_message_ids=[],
        global_attached_to_language=None,
        languages={
            Language.FI: LanguageBinding(
                language=Language.FI,
                image_message_ids=[],
                text_segments=[TextSegmentBinding(message_id=101, mode="message")],
                primary_message_id=101,
                primary_message_url="https://t.me/c/123/101",
                discussion_message_id=201,
            ),
            Language.SV: LanguageBinding(
                language=Language.SV,
                image_message_ids=[],
                text_segments=[TextSegmentBinding(message_id=102, mode="message")],
                primary_message_id=102,
                primary_message_url="https://t.me/c/123/102",
                discussion_message_id=202,
            ),
        },
        discussion_message_ids={101: 201, 102: 202},
    )

    result = apply_discussion_group_pinning(
        Config(bot_token="token", chat_id="-1001", channel_username="channel", discussion_group_id="-2001"),
        binding,
        [Language.FI, Language.SV],
        "first",
    )

    assert result.binding == binding
    assert result.pinning_failed is True