tests/test_telegram.py

from finbot.telegram import SeenUpdates, chunk_message, parse_update, redact


class TestRedact:
    """Space logs are world-readable on a public Space; tokens must not reach them."""

    TOKEN = "8984845302:AAEPEL6yzFjMxJcVKT5e_-D_4n3DGUnofzI"

    def test_removes_the_configured_token(self):
        text = f"GET https://api.telegram.org/bot{self.TOKEN}/getMe failed"
        assert self.TOKEN not in redact(text, self.TOKEN)

    def test_removes_a_token_shape_even_when_unknown(self):
        # e.g. a stale token still embedded in a cached exception string.
        text = "https://api.telegram.org/bot1234567:AAbbCC-_dd/sendMessage"
        assert "AAbbCC" not in redact(text, "")

    def test_keeps_the_useful_part(self):
        out = redact(f"bot{self.TOKEN}/getMe returned 401 Unauthorized", self.TOKEN)
        assert "401 Unauthorized" in out

    def test_handles_exceptions_and_none(self):
        assert redact(RuntimeError("boom"), "") == "boom"
        assert "None" in redact(None, "")

    def test_no_token_configured_is_safe(self):
        assert redact("nothing secret here", "") == "nothing secret here"


def message_update(update_id=1, **message):
    base = {"from": {"id": 42}, "chat": {"id": 99}}
    base.update(message)
    return {"update_id": update_id, "message": base}


class TestParseUpdate:
    def test_plain_text(self):
        parsed = parse_update(message_update(text="coffee 250"))
        assert parsed.user_id == 42
        assert parsed.chat_id == 99
        assert parsed.text == "coffee 250"
        assert parsed.command == ""

    def test_command_with_args(self):
        parsed = parse_update(message_update(text="/report last_month"))
        assert parsed.command == "report"
        assert parsed.args == "last_month"

    def test_bare_command(self):
        parsed = parse_update(message_update(text="/month"))
        assert parsed.command == "month"
        assert parsed.args == ""

    def test_command_is_lowercased(self):
        assert parse_update(message_update(text="/MONTH")).command == "month"

    def test_group_mention_is_stripped(self):
        parsed = parse_update(message_update(text="/month@FinBot"), bot_username="FinBot")
        assert parsed.command == "month"

    def test_mention_of_another_bot_is_ignored(self):
        assert parse_update(message_update(text="/month@OtherBot"), bot_username="FinBot") is None

    def test_photo_picks_the_largest_size(self):
        parsed = parse_update(
            message_update(
                photo=[
                    {"file_id": "small", "file_size": 100},
                    {"file_id": "large", "file_size": 9000},
                ],
                caption="lunch",
            )
        )
        assert parsed.file_id == "large"
        assert parsed.file_kind == "photo"
        assert parsed.mime_type == "image/jpeg"
        assert parsed.caption == "lunch"
        assert parsed.has_file

    def test_document(self):
        parsed = parse_update(
            message_update(
                document={
                    "file_id": "doc1",
                    "file_name": "statement.pdf",
                    "mime_type": "application/pdf",
                    "file_size": 2048,
                }
            )
        )
        assert parsed.file_kind == "document"
        assert parsed.file_name == "statement.pdf"
        assert parsed.file_size == 2048

    def test_edited_message_is_handled(self):
        update = {"update_id": 5, "edited_message": {"from": {"id": 42}, "chat": {"id": 99}, "text": "x"}}
        assert parse_update(update).text == "x"

    def test_non_message_updates_are_skipped(self):
        assert parse_update({"update_id": 1, "poll": {}}) is None

    def test_malformed_update_is_skipped(self):
        assert parse_update({"update_id": 1, "message": {"text": "hi"}}) is None
        assert parse_update({}) is None

    def test_no_file_means_no_file(self):
        assert not parse_update(message_update(text="hi")).has_file


class TestChunkMessage:
    def test_short_message_is_untouched(self):
        assert chunk_message("hello") == ["hello"]

    def test_splits_on_line_boundaries(self):
        text = "\n".join(f"line {i}" for i in range(500))
        chunks = chunk_message(text, limit=200)
        assert len(chunks) > 1
        assert all(len(c) <= 200 for c in chunks)
        # No content lost.
        assert "\n".join(chunks) == text

    def test_never_splits_mid_line(self):
        chunks = chunk_message("aaa\nbbb\nccc", limit=8)
        for chunk in chunks:
            for line in chunk.split("\n"):
                assert line in ("aaa", "bbb", "ccc")


class TestSeenUpdates:
    def test_first_time_is_new(self):
        seen = SeenUpdates()
        assert seen.add_if_new(1) is True

    def test_redelivery_is_rejected(self):
        seen = SeenUpdates()
        seen.add_if_new(1)
        assert seen.add_if_new(1) is False

    def test_evicts_oldest_beyond_capacity(self):
        seen = SeenUpdates(capacity=3)
        for i in range(5):
            seen.add_if_new(i)
        # 0 and 1 were evicted, so they read as new again.
        assert seen.add_if_new(0) is True
        assert seen.add_if_new(4) is False