Blame for tests/test_units.py

import asyncio
import datetime
from types import SimpleNamespace

import pytest
from telethon import TelegramClient, types
from telethon.sessions import MemorySession

from app import buttons as B
from app.common import UserError
from app.db import normalize_db_url
from app.channelref import parse_channel_ref
from app.handlers.panel import extract_content
from app.handlers.replace import parse_post_link, parse_replace_args
from app.linkswap import apply_span_replacements, classify_url, normalize_username, swap_url
from app.suggest import advance_plain, build_suggestion
from app.tgutil import (
    build_markup,
    button_url,
    classify_media,
    de_entities,
    is_callback_button,
    media_from_ref,
    pack_file_id,
    ser_entities,
    ser_markup,
    utf16_len,
)

from .fakes import FakeChannelClient, cb_btn, make_msg, markup, photo_media, url_btn
from .harness import App

NOW = datetime.datetime.now(datetime.timezone.utc)


def photo_msg(caption="cap", entities=None):
    photo = types.Photo(1, 2, b"ref", NOW, [types.PhotoSize("x", 100, 100, 999)], 2)
    return make_msg(5, caption, entities, media=types.MessageMediaPhoto(photo=photo))


def video_msg(gif=False):
    attrs = [types.DocumentAttributeVideo(3.0, 100, 100)]
    if gif:
        attrs.append(types.DocumentAttributeAnimated())
    doc = types.Document(9, 8, b"ref", NOW, "video/mp4", 1000, 2, attrs)
    return make_msg(6, "v", media=types.MessageMediaDocument(document=doc))


# ------------------------------------------------------------------ entities / markup
def test_entity_roundtrip_keeps_formatting_and_unknown_types():
    ents = [
        types.MessageEntityBold(0, 3),
        types.MessageEntityTextUrl(4, 2, "https://t.me/x?start=1"),
        types.MessageEntityPre(7, 3, "python"),
        types.MessageEntitySpoiler(11, 2),
        types.MessageEntityUrl(14, 5),  # auto entity: not stored
        types.MessageEntityFormattedDate(20, 4, NOW),  # unknown to us: kept byte-for-byte
    ]
    data = ser_entities(ents)
    assert [d["k"] for d in data] == ["bold", "url", "pre", "spoiler", "raw"]
    back = de_entities(data)
    assert [type(e).__name__ for e in back] == [
        "MessageEntityBold", "MessageEntityTextUrl", "MessageEntityPre", "MessageEntitySpoiler", "MessageEntityFormattedDate"
    ]
    assert back[1].url == "https://t.me/x?start=1" and back[2].language == "python"


def test_markup_roundtrip_keeps_other_bots_buttons_and_colours():
    style = types.KeyboardButtonStyle(bg_success=True)
    coloured = types.KeyboardButton(text="Go", type=types.InlineButtonTypeUrl("https://t.me/x"), style=style)
    mk = markup([coloured, cb_btn("👍 3", b"react:1")], [url_btn("Two", "https://t.me/y")])
    rows = ser_markup(mk)
    assert rows[0][0]["u"] == "https://t.me/x" and "s" in rows[0][0]
    assert rows[0][1]["raw"] and rows[1][0]["u"] == "https://t.me/y"
    rebuilt = build_markup(rows)
    b0, b1 = rebuilt.rows[0].buttons
    assert button_url(b0) == "https://t.me/x" and b0.style.bg_success
    assert is_callback_button(b1) and b1.type.data == b"react:1"


def test_utf16_len_counts_emoji_as_two():
    assert utf16_len("a😀") == 3


def test_apply_span_replacements_shifts_entities():
    ents = [types.MessageEntityBold(14, 3), types.MessageEntityItalic(0, 4), types.MessageEntityCode(5, 13)]
    text, out = apply_span_replacements("open t.me/abc now", ents, [(10, 13, "longer")])
    assert text == "open t.me/longer now"
    assert (out[0].offset, out[0].length) == (17, 3)  # after the change: moved by +3
    assert (out[1].offset, out[1].length) == (0, 4)  # before it: untouched
    assert (out[2].offset, out[2].length) == (5, 16)  # spanning it: grows


# -------------------------------------------------------------------- media handling
def test_extract_content_photo_video_gif_and_text():
    c = extract_content(photo_msg("hello", [types.MessageEntityBold(0, 5)]))
    assert c["media_kind"] == "photo" and c["text"] == "hello" and c["entities"][0]["k"] == "bold"
    assert c["media_file_id"]
    assert classify_media(video_msg()) == "video" and classify_media(video_msg(gif=True)) == "animation"
    t = extract_content(make_msg(1, "plain"))
    assert t["media_kind"] is None and t["media_file_id"] is None
    with pytest.raises(UserError):
        extract_content(make_msg(1, "   "))
    with pytest.raises(UserError):
        extract_content(photo_msg("x" * 1025))
    with pytest.raises(UserError):  # a poll can't be used in a post
        extract_content(make_msg(1, "q", media=types.MessageMediaPoll(types.Poll(1, [], types.TextWithEntities("q", []), None), types.PollResults())))


def test_link_preview_photo_is_not_mistaken_for_a_photo_post():
    page = types.WebPage(id=1, url="https://x.com", display_url="x.com", hash=0, photo=types.Photo(1, 2, b"r", NOW, [], 2))
    m = make_msg(2, "look https://x.com", media=types.MessageMediaWebPage(webpage=page))
    assert classify_media(m) is None
    assert extract_content(m)["media_kind"] is None


def test_bot_file_id_can_be_reused_when_sending():
    m = photo_msg()
    fid = pack_file_id(m)
    assert fid.startswith("tl:")

    async def go(blank):
        client = TelegramClient(MemorySession(), 1, "h")
        # what send_file / edit_message do with the media object
        _, media, _ = await client._file_to_media(media_from_ref(fid, blank_reference=blank))
        return media

    assert isinstance(asyncio.run(go(False)), types.InputMediaPhoto)
    blank = asyncio.run(go(True))
    assert blank.id.file_reference == b"" and blank.id.id == 1 and blank.id.access_hash == 2
    doc_ref = pack_file_id(video_msg())
    assert media_from_ref(doc_ref).document.id == 9


# --------------------------------------------------------------------------- parsing
def test_button_parsing_and_limits():
    assert B.parse_buttons_text("A - t.me/x | B - https://y.com/?a=1\nC — https://z.com") == [
        [{"t": "A", "u": "https://t.me/x"}, {"t": "B", "u": "https://y.com/?a=1"}],
        [{"t": "C", "u": "https://z.com"}],
    ]
    for bad in ("no link here", "- https://x.com", "A - htp://x"):
        with pytest.raises(ValueError):
            B.parse_buttons_text(bad)
    with pytest.raises(ValueError):
        B.parse_buttons_text(" | ".join(f"b{i} - https://x.com/{i}" for i in range(9)))


def test_move_button_edges():
    rows = [[{"t": "a", "u": "u"}, {"t": "b", "u": "u"}], [{"t": "c", "u": "u"}]]
    assert B.move_button(rows, 0, 0, "l")[0] == rows and B.move_button(rows, 0, 0, "u")[0] == rows
    r, nr, nc = B.move_button(rows, 1, 0, "u")  # keeps its column when it can
    assert [[b["t"] for b in x] for x in r] == [["c", "a", "b"]] and (nr, nc) == (0, 0)
    r, nr, nc = B.move_button(rows, 0, 1, "d")
    assert [[b["t"] for b in x] for x in r] == [["a"], ["c", "b"]] and (nr, nc) == (1, 1)
    r, nr, nc = B.move_button([[{"t": "a"}, {"t": "b"}]], 0, 1, "d")  # last row with company -> new row
    assert [[b["t"] for b in x] for x in r] == [["a"], ["b"]] and (nr, nc) == (1, 0)
    assert B.delete_button([[{"t": "a"}]], 0, 0) == []


def test_replace_args_and_refs():
    old, new, f = parse_replace_args("@Bharath goku --last 50 --posts --channel @ch")
    assert (old, new, f["last"], f["posts"], f["channel"]) == ("Bharath", "goku", 50, True, "@ch")
    for bad in ("@a", "@a @a", "@a @b --wat", "@a @b --last x", "@a b!d"):
        with pytest.raises(ValueError):
            parse_replace_args(bad)
    assert normalize_username("https://t.me/foo_bar/") == "foo_bar" and normalize_username("@x y") is None
    assert parse_channel_ref("@animech") == "animech" and parse_channel_ref("https://t.me/c/123456/7") == 123456
    assert parse_channel_ref("-1001234567890") == 1234567890 and parse_channel_ref("???") is None
    assert parse_post_link("https://t.me/c/1723201445/9084") == (1723201445, 9084)
    assert parse_post_link("https://t.me/animech/9084") == ("animech", 9084)


def test_replace_adopts_a_linked_private_post_then_updates_its_media(tmp_path):
    async def run():
        app = App(tmp_path)
        await app.start()
        channel_id, message_id = 1723201445, 9084
        old = make_msg(message_id, "old caption", media=photo_media(10))
        app.tg.add_channel(FakeChannelClient({message_id: old}, cid=channel_id, title="Database Channel"))
        await app.db.save_channel(channel_id, 5, "Database Channel", None, 1)
        replacement = make_msg(1, "new caption", media=photo_media(20))

        await app.text(f"/replace https://t.me/c/{channel_id}/{message_id}", reply=replacement)

        saved = await app.db.find_by_message(channel_id, message_id)
        assert saved is not None and saved.source == "adopted"
        assert saved.text == "new caption" and saved.media_file_id == pack_file_id(replacement)
        assert app.tg.edits[-1][1] == message_id
        assert "Replaced the media" in app.out.last_text
        await app.db.close()

    asyncio.run(run())


def test_url_helpers():
    assert classify_url("https://T.me/Bharath?start=x", "bharath") == "link"
    assert classify_url("https://t.me/bharath/12", "bharath") == "post"
    assert classify_url("https://t.me/bharath_bot", "bharath") is None
    assert classify_url("https://example.com/bharath", "bharath") is None
    assert swap_url("http://telegram.me/Bharath?start=A_b-C", "bharath", "goku") == "http://telegram.me/goku?start=A_b-C"


def test_db_url_normalisation():
    url, args = normalize_db_url("postgres://u:p@ep-x-pooler.eu.neon.tech/db?sslmode=require&channel_binding=require")
    assert url.startswith("postgresql+asyncpg://") and "sslmode" not in url and "channel_binding" not in url
    assert args == {"ssl": True, "statement_cache_size": 0} and "prepared_statement_cache_size=0" in url
    assert normalize_db_url("sqlite:///data/x.db")[0] == "sqlite+aiosqlite:///data/x.db"


# ------------------------------------------------------------------------ suggestions
def test_suggestion_details():
    post = SimpleNamespace(
        id=1, text="Episodes 01 to 20 are here 👇", entities=[{"k": "bold", "o": 0, "l": 8}],
        buttons=[[{"t": "Download Episodes 01 to 20", "u": "https://x"}, {"t": "Join", "u": "https://j"}],
                 [{"t": "👍 2", "raw": "AAAA"}]],
        media_kind="photo", media_file_id="FID",
    )
    s = build_suggestion(post)
    assert s.text == "Episodes 21 to 40 are here 👇" and s.entities[0]["l"] == 8
    assert s.buttons == [[{"t": "Download Episodes 21 to 40", "u": None}, {"t": "Join", "u": "https://j"}]]
    assert s.pending == [(0, 0)] and s.media_file_id == "FID"  # reactions from another bot are dropped
    assert advance_plain("no numbers") is None
    assert build_suggestion(SimpleNamespace(id=2, text="", entities=[], buttons=[[{"t": "Join", "u": "u"}]], media_kind=None, media_file_id=None)) is None


# ------------------------------------------------------- channel registration handler
def test_add_channel_by_forward_and_rights(tmp_path):
    async def run():
        app = App(tmp_path)
        await app.start()
        ch = types.Channel(id=77, title="Fwd Channel", photo=types.ChatPhotoEmpty(), date=NOW, broadcast=True,
                           access_hash=1234, username="fwdch")
        msg = SimpleNamespace(raw_text="", entities=[], media=None, grouped_id=None, forward=SimpleNamespace(chat=ch))
        await app.text("/addchannel")
        await app.text("", message=msg)
        assert "Added" in app.out.last_text
        saved = await app.db.get_channel(77)
        assert (saved.access_hash, saved.username, saved.active) == (1234, "fwdch", True)

        # a "min" channel without access details is refused with an explanation
        mini = types.Channel(id=78, title="Min", photo=types.ChatPhotoEmpty(), date=NOW, broadcast=True, min=True)
        await app.text("/addchannel")
        await app.text("", message=SimpleNamespace(raw_text="", entities=[], media=None, grouped_id=None, forward=SimpleNamespace(chat=mini)))
        assert "access details" in app.out.last_text
        assert await app.db.get_channel(78) is None
        await app.db.close()

    asyncio.run(run())


def test_media_post_publish_uses_send_file_and_caption_edit(tmp_path):
    async def run():
        app = App(tmp_path)
        await app.start()
        await app.db.save_channel(1, 5, "C", None, 1)
        p = await app.db.create_post(channel_id=1, status="draft", source="bot", text="cap", entities=[], media_kind="photo",
                                     media_file_id=pack_file_id(photo_msg()), buttons=[[{"t": "Go", "u": "https://x.com"}]], created_by=1)
        await app.press(f"pby:{p.id}")
        kind, peer, file, kw = app.tg.sent[-1]
        assert kind == "file" and isinstance(file, types.MessageMediaPhoto) and kw["caption"] == "cap"
        # editing the caption of the live media post goes through messages.editMessage
        await app.press(f"et:{p.id}")
        await app.text("new caption")
        req = app.tg.edits[-1][2]
        assert req.message == "new caption" and len(req.reply_markup.rows[0].buttons) == 1
        # replacing the media uses Telegram's file-edit path
        await app.press(f"em:{p.id}")
        await app.text("nope")
        assert "Send a photo" in app.out.last_text
        await app.db.close()

    asyncio.run(run())