Spaces:
Paused
Paused
File size: 8,115 Bytes
de342dc | 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 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 | """Checking and repairing the buttons a channel really shows."""
import asyncio
import pytest
from app.buttons_sync import ButtonsNotShown, ensure_markup, fix_buttons, link_urls, scan_buttons
from app.tgutil import build_markup, button_url, peer_of
from .dbutil import db_url_for, fresh_db
from .fakes import FakeChannelClient, make_msg, markup, rpc, url_btn
from .harness import App, norm
A = "https://t.me/goku?start=A"
B = "https://t.me/goku?start=B"
ROWS = [[{"t": "Get", "u": A}], [{"t": "More", "u": B}]]
def shown(chan, i):
m = chan.msgs[i]
return [button_url(b) for r in (m.reply_markup.rows if m.reply_markup else []) for b in r.buttons]
class Peer:
id, access_hash = 1, 5
def run(coro):
return asyncio.run(coro)
def test_link_urls():
assert link_urls(None) == []
assert link_urls(markup([url_btn("x", A), url_btn("y", B)])) == [A, B]
def test_ensure_markup_in_every_situation():
async def go():
want = build_markup(ROWS)
peer = peer_of(Peer)
chan = FakeChannelClient({1: make_msg(1, "ok", markup=want)})
assert await ensure_markup(chan, peer, 1, want, pause=0) == "fine"
assert chan.edits == []
chan = FakeChannelClient({1: make_msg(1, "no keyboard")}) # simply missing: one edit is enough
assert await ensure_markup(chan, peer, 1, want, pause=0) == "set"
assert shown(chan, 1) == [A, B] and len(chan.edits) == 1 and chan.msgs[1].message == "no keyboard"
chan = FakeChannelClient({1: make_msg(1, "hidden")}) # Telegram believes the keyboard is there already
chan.msgs[1]._hidden = want
assert await ensure_markup(chan, peer, 1, want, pause=0) == "nudged"
assert shown(chan, 1) == [A, B] and len(chan.edits) == 2
chan = FakeChannelClient({})
with pytest.raises(LookupError):
await ensure_markup(chan, peer, 1, want, pause=0)
chan = FakeChannelClient({1: make_msg(1, "x")})
chan.drop_all_markup = True # edits are accepted but the keyboard never appears
with pytest.raises(ButtonsNotShown):
await ensure_markup(chan, peer, 1, want, pause=0)
chan = FakeChannelClient({1: make_msg(1, "x")}) # nothing to show -> nothing to do
assert await ensure_markup(chan, peer, 1, None, pause=0) == "fine" and chan.requests == []
run(go())
async def seeded(tmp_path):
"""Channel + database as /repost used to leave them: the database knows the buttons, the channel may not show them."""
db = await fresh_db(db_url_for(tmp_path))
await db.save_channel(1, 5, "Test", "testch", 99)
ch = await db.get_channel(1)
want = build_markup(ROWS)
msgs = {
10: make_msg(10, "fine", markup=want),
11: make_msg(11, "missing"),
12: make_msg(12, "hidden"),
13: make_msg(13, "other link", markup=markup([url_btn("Get", "https://t.me/other?start=Z")])),
15: make_msg(15, "no buttons saved"),
}
chan = FakeChannelClient(msgs)
chan.msgs[12]._hidden = want
for i in (10, 11, 12, 13, 14): # 14 was deleted by hand
await db.create_post(channel_id=1, message_id=i, status="sent", source="bot", text="t", entities=[], buttons=ROWS)
await db.create_post(channel_id=1, message_id=15, status="sent", source="bot", text="t", entities=[], buttons=[])
await db.create_post(channel_id=1, message_id=None, status="draft", source="bot", text="draft", entities=[], buttons=ROWS)
return db, ch, chan
def test_scan_finds_exactly_the_posts_with_wrong_buttons(tmp_path):
async def go():
db, ch, chan = await seeded(tmp_path)
scan = await scan_buttons(chan, ch, await db.sent_posts(1))
assert scan.total == 5 # 10..14 have saved link buttons; 15 and the draft do not count
assert scan.fine == 1
assert sorted(p.message_id for p, _, _ in scan.bad) == [11, 12, 13]
assert [p.message_id for p in scan.gone] == [14]
live13 = next(live for p, live, _ in scan.bad if p.message_id == 13)
assert live13 == ["https://t.me/other?start=Z"]
await db.close()
run(go())
def test_fix_puts_the_saved_buttons_back(tmp_path):
async def go():
db, ch, chan = await seeded(tmp_path)
scan = await scan_buttons(chan, ch, await db.sent_posts(1))
res = await fix_buttons(chan, ch, scan.bad, delay=0, pause=0)
assert (res.fixed, res.nudged, res.already, res.gone, res.failed, res.aborted) == (3, 1, 0, 0, [], None)
for i in (10, 11, 12, 13):
assert shown(chan, i) == [A, B]
assert chan.msgs[11].message == "missing" # texts untouched
again = await scan_buttons(chan, ch, await db.sent_posts(1))
assert again.bad == [] and again.fine == 4 and len(again.gone) == 1
await db.close()
run(go())
def test_fix_gives_up_after_several_identical_refusals(tmp_path):
async def go():
db = await fresh_db(db_url_for(tmp_path))
await db.save_channel(1, 5, "Test", "testch", 99)
ch = await db.get_channel(1)
chan = FakeChannelClient({i: make_msg(i, "x") for i in range(1, 9)})
chan.fail["EditMessageRequest"] = rpc("MessageIdInvalidError")
for i in range(1, 9):
await db.create_post(channel_id=1, message_id=i, status="sent", source="bot", text="x", entities=[], buttons=ROWS)
scan = await scan_buttons(chan, ch, await db.sent_posts(1))
res = await fix_buttons(chan, ch, scan.bad, delay=0, pause=0)
assert res.fixed == 0 and len(res.failed) == 5 and res.aborted == "MessageIdInvalidError"
await db.close()
run(go())
# ------------------------------------------------------------------------------------------ the button in My posts
def test_check_buttons_in_my_posts(tmp_path):
async def go():
app = App(tmp_path)
await app.start()
db, ch, chan = await seeded_app(app)
await app.press("pc:1:0")
assert "Check buttons" in str(app.out.last_buttons())
await app.press("pk:1")
t = norm(app.out.last_text)
assert "checked 5 post(s)" in t and "1 show their buttons correctly" in t and "3 are saved with buttons" in t
assert "#11" in t and "#12" in t and "#13" in t and "1 saved post(s) no longer exist" in t
assert "Fix 3 post(s)" in str(app.out.last_buttons())
await app.press(app.out.callback_data("Fix 3"))
t = norm(app.out.last_text)
assert "put the saved buttons back on 3 post(s)" in t and "1 of them needed a second edit" in t
for i in (10, 11, 12, 13):
assert shown(app.tg, i) == [A, B]
await app.press("pk:1")
assert "Everything matches" in norm(app.out.last_text) and "Fix" not in str(app.out.last_buttons())
await app.db.close()
run(go())
async def seeded_app(app):
await app.db.save_channel(1, 5, "Anime Channel", "animech", 1)
want = build_markup(ROWS)
app.tg.msgs.update(
{
10: make_msg(10, "fine", markup=want),
11: make_msg(11, "missing"),
12: make_msg(12, "hidden"),
13: make_msg(13, "other link", markup=markup([url_btn("Get", "https://t.me/other?start=Z")])),
15: make_msg(15, "no buttons saved"),
}
)
app.tg.msgs[12]._hidden = want
for i in (10, 11, 12, 13, 14):
await app.db.create_post(channel_id=1, message_id=i, status="sent", source="bot", text="t", entities=[], buttons=ROWS)
return app.db, None, app.tg
def test_check_buttons_is_for_the_owner_only(tmp_path):
async def go():
app = App(tmp_path, owner=1, admins=frozenset({2}))
await app.start()
await seeded_app(app)
app.uid = 2 # an admin who may make posts but is not the owner
await app.press("pc:1:0")
assert "Check buttons" not in str(app.out.last_buttons())
await app.press("pk:1")
assert "Only the bot owner" in app.out.last_text
await app.press("pkf:1")
assert "Only the bot owner" in app.out.last_text
await app.db.close()
run(go())
|