Spaces:
Paused
Paused
File size: 6,089 Bytes
61d6e2a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 | import asyncio
from telethon import types
from app.linkswap import compute_message_changes
from app.replace_engine import ScanOptions, apply_channel, scan_channel, undo_batch
from app.tgutil import button_url, is_callback_button
from .dbutil import fresh_db, db_url_for
from .fakes import FakeClient, cb_btn, make_msg, markup, rpc, url_btn
OLD = "https://t.me/bharath?start=BBB"
def build():
text3 = "Hi 😀 https://t.me/bharath?start=CCC done"
msgs = {
1: make_msg(1, "Get it now", [types.MessageEntityTextUrl(0, 3, "https://t.me/Bharath?start=AAA")]),
2: make_msg(
2,
"Episodes",
markup=markup([url_btn("Ep 1-20", OLD), cb_btn("👍 3", b"r1")], [url_btn("Join", "https://t.me/other")]),
out=True,
),
3: make_msg(3, text3, [types.MessageEntityBold(37, 4)]),
4: make_msg(4, "see https://t.me/bharath_bot?start=1"),
5: make_msg(5, "https://t.me/bharath/123"),
7: types.MessageEmpty(id=7, peer_id=types.PeerChannel(1)),
}
return msgs, text3
async def setup(tmp_path, msgs, pts=8):
db = await fresh_db(db_url_for(tmp_path))
await db.save_channel(1, 5, "Test", None, 99)
ch = await db.get_channel(1)
return db, ch, FakeClient(msgs, pts=pts)
def test_scan_apply_undo(tmp_path):
async def run():
msgs, text3 = build()
db, ch, client = await setup(tmp_path, msgs)
opts = ScanOptions("bharath", "goku")
scan = await scan_channel(client, ch, opts, can_edit_others=True)
assert set(scan.infos) == {1, 2, 3}
assert scan.skipped_post_links == 1 # t.me/bharath/123 is a channel post, not a bot link
assert scan.foreign == 2 and scan.infos[2].mine and scan.infos[2].callbacks
await db.create_batch("b1", "bharath", "goku", 99)
res = await apply_channel(client, db, scan, opts, "b1", 99, edit_delay=0)
assert (res.edited, res.failed) == (3, [])
# hyperlink: only the username changed
assert msgs[1].entities[0].url == "https://t.me/goku?start=AAA"
# buttons: URL swapped, reaction button + other link untouched, text untouched
rows = msgs[2].reply_markup.rows
assert button_url(rows[0].buttons[0]) == "https://t.me/goku?start=BBB"
assert is_callback_button(rows[0].buttons[1])
assert button_url(rows[1].buttons[0]) == "https://t.me/other"
assert msgs[2].message == "Episodes"
# typed link after an emoji: text swapped and the following entity moved with it
assert msgs[3].message == "Hi 😀 https://t.me/goku?start=CCC done"
assert (msgs[3].entities[0].offset, msgs[3].entities[0].length) == (34, 4)
# untouched
assert msgs[4].message == "see https://t.me/bharath_bot?start=1"
assert msgs[5].message == "https://t.me/bharath/123"
# adopted into the database
p = await db.find_by_message(1, 3)
assert p.source == "adopted" and "goku" in p.text
assert (await db.find_by_message(1, 2)).buttons[0][0]["u"].endswith("goku?start=BBB")
# undo restores everything
ures = await undo_batch(client, db, "b1", edit_delay=0)
assert ures.restored == 3 and not ures.failed
assert msgs[1].entities[0].url == "https://t.me/Bharath?start=AAA"
assert msgs[3].message == text3 and msgs[3].entities[0].offset == 37
assert button_url(msgs[2].reply_markup.rows[0].buttons[0]) == OLD
assert is_callback_button(msgs[2].reply_markup.rows[0].buttons[1])
assert (await db.last_batch()) is None # batch marked as undone
await db.close()
asyncio.run(run())
def test_only_own_posts_without_edit_right(tmp_path):
async def run():
msgs, _ = build()
db, ch, client = await setup(tmp_path, msgs)
opts = ScanOptions("bharath", "goku")
scan = await scan_channel(client, ch, opts, can_edit_others=False)
assert set(scan.infos) == {2} and scan.not_editable == 2
await db.create_batch("b2", "bharath", "goku", 99)
res = await apply_channel(client, db, scan, opts, "b2", 99, edit_delay=0)
assert res.edited == 1 and len(client.edits) == 1
await db.close()
asyncio.run(run())
def test_failures_are_reported_not_fatal(tmp_path):
async def run():
msgs, _ = build()
db, ch, client = await setup(tmp_path, msgs)
client.fail[1] = rpc("MessageAuthorRequiredError")
opts = ScanOptions("bharath", "goku")
scan = await scan_channel(client, ch, opts)
await db.create_batch("b3", "bharath", "goku", 99)
res = await apply_channel(client, db, scan, opts, "b3", 99, edit_delay=0)
assert res.edited == 2 and res.failed == [(1, "MessageAuthorRequiredError")]
await db.close()
asyncio.run(run())
def test_scan_without_counter_falls_back_to_probing(tmp_path):
async def run():
msgs, _ = build()
db, ch, client = await setup(tmp_path, msgs, pts=None)
scan = await scan_channel(client, ch, ScanOptions("bharath", "goku"))
assert set(scan.infos) == {1, 2, 3}
await db.close()
asyncio.run(run())
def test_post_links_can_be_included():
m = make_msg(5, "https://t.me/bharath/123")
assert compute_message_changes(m, "bharath", "goku")[0] is None
ch, skipped = compute_message_changes(m, "bharath", "goku", include_posts=True)
assert ch.new_text == "https://t.me/goku/123"
def test_tg_scheme_and_hyperlink_and_text_flags():
m = make_msg(
6,
"open tg://resolve?domain=bharath&start=x",
[types.MessageEntityTextUrl(0, 4, "tg://resolve?domain=Bharath&start=y")],
)
ch, _ = compute_message_changes(m, "bharath", "goku")
assert ch.new_text == "open tg://resolve?domain=goku&start=x"
assert ch.new_entities[0].url == "tg://resolve?domain=goku&start=y"
assert ch.n_typed == 1 and ch.n_links == 1
ch2, _ = compute_message_changes(m, "bharath", "goku", include_typed=False)
assert ch2.new_text == m.message and ch2.n_typed == 0
|